fix(topic-data): fix Protobuf deserializer (tchiotludo#640)

Fix tchiotludo#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 <[email protected]>
sajiljosephs · Mar 18, 2021 · 9d3455d · 9d3455d
1 parent 7216a99
commit 9d3455d
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 4 deletions.
diff --git 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
@@ -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();