Merge pull request #672 from YotpoLtd/decode-null-messages-using-avro…

…-deserializer Send null messages to kafka avro deserializer
pinterest · Apr 14, 2019 · 985eccc · 985eccc
2 parents 7d78f3b + e11ccb8
commit 985eccc
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 2 deletions.
diff --git a/src/main/java/com/pinterest/secor/common/SecorSchemaRegistryClient.java b/src/main/java/com/pinterest/secor/common/SecorSchemaRegistryClient.java
@@ -56,9 +56,14 @@ protected void init(SecorConfig config) {
     }
 
     public GenericRecord decodeMessage(String topic, byte[] message) {
+        if (message.length == 0) {
+            message = null;
+        }
         GenericRecord record = (GenericRecord) deserializer.deserialize(topic, message);
-        Schema schema = record.getSchema();
-        schemas.put(topic, schema);
+        if (record != null) {
+            Schema schema = record.getSchema();
+            schemas.put(topic, schema);
+        }
         return record;
     }
 

diff --git a/src/test/java/com/pinterest/secor/common/SecorSchemaRegistryClientTest.java b/src/test/java/com/pinterest/secor/common/SecorSchemaRegistryClientTest.java
@@ -99,5 +99,8 @@ public void testDecodeMessage() throws Exception {
         assertEquals(output.get("data_field_1"), 1);
         assertTrue(StringUtils.equals((output.get("data_field_2")).toString(), "hello"));
         assertEquals(output.get("timestamp"), 1467176316L);
+
+        output = secorSchemaRegistryClient.decodeMessage("test-avr-topic", new byte[0]);
+        assertNull(output);
     }
 }