fix(topicdata): support default values in deserializer (#776)

close #761
tchiotludo · Oct 24, 2021 · a5a7eb8 · a5a7eb8
1 parent 675a6bf
commit a5a7eb8
Show file tree

Hide file tree

Showing 2 changed files with 31 additions and 8 deletions.
diff --git a/src/main/java/org/akhq/utils/AvroSerializer.java b/src/main/java/org/akhq/utils/AvroSerializer.java
@@ -50,20 +50,22 @@ public class AvroSerializer {
             .appendOptional(DateTimeFormatter.ofPattern("XXX"))
             .toFormatter();
 
-    public static GenericRecord recordSerializer(Map<String, ?> record, Schema schema) {
+    public static GenericRecord recordSerializer(Map<String, Object> record, Schema schema) {
         GenericRecord returnValue = new GenericData.Record(schema);
         schema
             .getFields()
-            .forEach(f -> returnValue.put(
-                    f.name(),
-                    AvroSerializer.objectSerializer(record.get(f.name()), schema.getField(f.name()).schema())
-                )
-            );
+            .forEach(field -> {
+                Object fieldValue = record.getOrDefault(field.name(), field.defaultVal());
+                returnValue.put(field.name(), AvroSerializer.objectSerializer(fieldValue, field.schema()));
+            });
         return returnValue;
     }
 
     @SuppressWarnings("unchecked")
     private static Object objectSerializer(Object value, Schema schema) {
+        if (value == org.apache.avro.JsonProperties.NULL_VALUE) {
+            return null;
+        }
         LogicalType logicalType = schema.getLogicalType();
         Schema.Type primitiveType = schema.getType();
         if (logicalType != null) {
@@ -92,7 +94,7 @@ private static Object objectSerializer(Object value, Schema schema) {
                 case MAP:
                     return AvroSerializer.mapSerializer((Map<String, ?>) value, schema);
                 case RECORD:
-                    return AvroSerializer.recordSerializer((Map<String, ?>) value, schema);
+                    return AvroSerializer.recordSerializer((Map<String, Object>) value, schema);
                 case ENUM:
                     return new GenericData.EnumSymbol(schema, value.toString());
                 case ARRAY:

diff --git a/src/test/java/org/akhq/utils/AvroDeserializerTest.java b/src/test/java/org/akhq/utils/AvroDeserializerTest.java
@@ -19,7 +19,6 @@
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
-import static org.junit.jupiter.api.Assertions.assertEquals;
 
 class AvroDeserializerTest {
     private static Schema fieldsToSchema(String s) {
@@ -159,4 +158,26 @@ void testCompleteObject() {
 
         genericTest(expected, type);
     }
+
+    @Test
+    void testDefaultValue() {
+        String type = "{"
+            + "\"name\": \"root\","
+            + "\"type\": \"record\","
+            + "\"fields\": ["
+            + "    {\"name\": \"stringField\", \"type\": [\"null\", \"string\"], \"default\": null},"
+            + "    {\"name\": \"arrayField\", \"type\": {\"type\": \"array\", \"items\": \"double\"}, \"default\": []}"
+            + "    ]"
+            + "}";
+        Schema schema = new Schema.Parser().parse(type);
+
+        GenericRecord expectedRecord = AvroSerializer.recordSerializer(Map.of(), schema);
+        assert new GenericData().validate(schema, expectedRecord);
+
+        Map<String, Object> result = AvroDeserializer.recordDeserializer(expectedRecord);
+        Map<String, Object> defaultValues = new HashMap<>();
+        defaultValues.put("stringField", null);
+        defaultValues.put("arrayField", List.of());
+        assertThat(result, is(defaultValues));
+    }
 }