From 9d3455dfebdc18906581dd235f3df12de96a52c9 Mon Sep 17 00:00:00 2001
From: Taisiia Goltseva <xakassi@mail.ru>
Date: Thu, 18 Mar 2021 23:28:05 +0400
Subject: [PATCH] fix(topic-data): fix Protobuf deserializer (#640)

Fix #621

When configure topic regexes for Protobuf deserialization, topic keys maybe not in Protobuf format, in that case key-message-type can be omitted. The same for value-message-type.
But there is a bug in the code, it shows a warning "Protobuf deserialization is configured for topic [%s], but message type is not specified neither for a key, nor for a value.", when message type is configured only for a key or only for a value.

It should be show only when both key and value message types are not configured.

Co-authored-by: Taisiia Goltseva <tado0618@netcracker.com>
---
 .../akhq/utils/ProtobufToJsonDeserializer.java |  9 +++++++--
 .../utils/ProtobufToJsonDeserializerTest.java  | 18 ++++++++++++++++--
 2 files changed, 23 insertions(+), 4 deletions(-)

diff --git a/src/main/java/org/akhq/utils/ProtobufToJsonDeserializer.java b/src/main/java/org/akhq/utils/ProtobufToJsonDeserializer.java
index 016cd8f4e..59dfec214 100644
--- a/src/main/java/org/akhq/utils/ProtobufToJsonDeserializer.java
+++ b/src/main/java/org/akhq/utils/ProtobufToJsonDeserializer.java
@@ -117,14 +117,19 @@ public String deserialize(String topic, byte[] buffer, boolean isKey) {
             log.debug("Protobuf deserialization config is not found for topic [{}]", topic);
             return null;
         }
+
+        if (matchingConfig.getValueMessageType() == null && matchingConfig.getKeyMessageType() == null) {
+            throw new SerializationException(String.format("Protobuf deserialization is configured for topic [%s], " +
+                    "but message type is not specified neither for a key, nor for a value.", topic));
+        }
+
         String messageType = matchingConfig.getValueMessageType();
         if (isKey) {
             messageType = matchingConfig.getKeyMessageType();
         }
 
         if (messageType == null) {
-            throw new SerializationException(String.format("Protobuf deserialization is configured for topic [%s], " +
-                    "but message type is not specified neither for a key, nor for a value.", topic));
+            return null;
         }
 
         String result;
diff --git a/src/test/java/org/akhq/utils/ProtobufToJsonDeserializerTest.java b/src/test/java/org/akhq/utils/ProtobufToJsonDeserializerTest.java
index 59cb3872e..9d7aa4cc4 100644
--- a/src/test/java/org/akhq/utils/ProtobufToJsonDeserializerTest.java
+++ b/src/test/java/org/akhq/utils/ProtobufToJsonDeserializerTest.java
@@ -55,13 +55,19 @@ private void createTopicProtobufDeserializationMapping() throws URISyntaxExcepti
         filmTopicsMapping.setDescriptorFileBase64(base64FilmDescriptor);
         filmTopicsMapping.setValueMessageType("Film");
 
+        // Do not specify message type neither for a key, nor for a value
+        TopicsMapping incorrectTopicsMapping = new TopicsMapping();
+        incorrectTopicsMapping.setTopicRegex("incorrect.*");
+        String base64IncorrectDescriptor = encodeDescriptorFileToBase64("film.desc");
+        incorrectTopicsMapping.setDescriptorFileBase64(base64IncorrectDescriptor);
+
         TopicsMapping complexObjectTopicsMapping = new TopicsMapping();
         complexObjectTopicsMapping.setTopicRegex("complex.*");
         complexObjectTopicsMapping.setDescriptorFile("complex.desc");
         complexObjectTopicsMapping.setValueMessageType("Complex");
 
         protobufDeserializationTopicsMapping.setTopicsMapping(
-                Arrays.asList(albumTopicsMapping, filmTopicsMapping, complexObjectTopicsMapping));
+                Arrays.asList(albumTopicsMapping, filmTopicsMapping, complexObjectTopicsMapping, incorrectTopicsMapping));
     }
 
     private String encodeDescriptorFileToBase64(String descriptorFileName) throws URISyntaxException, IOException {
@@ -149,10 +155,18 @@ public void deserializeForNotMatchingTopic() {
 
     @Test
     public void deserializeForKeyWhenItsTypeNotSet() {
+        ProtobufToJsonDeserializer protobufToJsonDeserializer = new ProtobufToJsonDeserializer(protobufDeserializationTopicsMapping);
+        final byte[] binaryFilm = filmProto.toByteArray();
+        String decodedFilm = protobufToJsonDeserializer.deserialize("film.topic.name", binaryFilm, true);
+        assertNull(decodedFilm);
+    }
+
+    @Test
+    public void deserializeWhenTypeNotSetForKeyAndValue() {
         ProtobufToJsonDeserializer protobufToJsonDeserializer = new ProtobufToJsonDeserializer(protobufDeserializationTopicsMapping);
         final byte[] binaryFilm = filmProto.toByteArray();
         Exception exception = assertThrows(RuntimeException.class, () -> {
-            protobufToJsonDeserializer.deserialize("film.topic.name", binaryFilm, true);
+            protobufToJsonDeserializer.deserialize("incorrect.topic.name", binaryFilm, true);
         });
         String expectedMessage = "message type is not specified neither for a key, nor for a value";
         String actualMessage = exception.getMessage();