CC-7399: Handle null value if schema-less record has a map with a null value. (#220)

Author: avocader

kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/convert/BigQueryRecordConverter.java

@@ -110,22 +110,22 @@ private Object convertSchemalessRecord(Object value) {
     }
     if (value instanceof Map) {
       return
-        ((Map<Object, Object>) value).entrySet().stream().collect(
-                Collectors.toMap(
-                        entry -> {
-                          if (!(entry.getKey() instanceof String)) {
-                            throw new ConversionConnectException(
-                                    "Failed to convert record to bigQuery format: " +
-                                    "Map objects in absence of schema needs to have string value keys. ");
-                          }
-                          return entry.getKey();
-                        },
-                        entry -> convertSchemalessRecord(entry.getValue())
-                )
-        );
+          ((Map<Object, Object>) value)
+              .entrySet()
+              .stream()
+              .collect(HashMap::new,
+                  (m, e) -> {
+                    if (!(e.getKey() instanceof String)) {
+                      throw new ConversionConnectException(
+                          "Failed to convert record to bigQuery format: " +
+                              "Map objects in absence of schema needs to have string value keys. ");
+                    }
+                    m.put(e.getKey(), convertSchemalessRecord(e.getValue()));
+                  },
+                  HashMap::putAll);
     }
     throw new ConversionConnectException("Unsupported class " + value.getClass() +
-            " found in schemaless record data. Can't convert record to bigQuery format");
+        " found in schemaless record data. Can't convert record to bigQuery format");
   }
 
   @SuppressWarnings("unchecked")

kcbq-connector/src/test/java/com/wepay/kafka/connect/bigquery/convert/BigQueryRecordConverterTest.java

@@ -29,6 +29,7 @@
 
 import org.apache.kafka.connect.sink.SinkRecord;
 
+import org.junit.Assert;
 import org.junit.Test;
 
 import java.nio.ByteBuffer;
@@ -604,6 +605,37 @@ public void testInvalidMapSchemaless() {
             new BigQueryRecordConverter(SHOULD_CONVERT_DOUBLE).convertRecord(kafkaConnectRecord, KafkaSchemaRecordType.VALUE);
   }
 
+  @Test
+  public void testInvalidMapSchemalessNullValue() {
+    Map kafkaConnectMap = new HashMap<Object, Object>(){{
+      put("f1", "abc");
+      put("f2", "abc");
+      put("f3", null);
+    }};
+
+    SinkRecord kafkaConnectRecord = spoofSinkRecord(null, kafkaConnectMap);
+    Map<String, Object> stringObjectMap = new BigQueryRecordConverter(SHOULD_CONVERT_DOUBLE).convertRecord(kafkaConnectRecord);
+    Assert.assertEquals(kafkaConnectMap, stringObjectMap
+    );
+  }
+
+  @Test
+  public void testInvalidMapSchemalessNestedMapNullValue() {
+    Map kafkaConnectMap = new HashMap<Object, Object>(){{
+      put("f1", "abc");
+      put("f2", "abc");
+      put("f3", new HashMap<Object, Object>() {{
+        put("f31", "xyz");
+        put("f32", null);
+      }});
+    }};
+
+    SinkRecord kafkaConnectRecord = spoofSinkRecord(null, kafkaConnectMap);
+    Map<String, Object> stringObjectMap = new BigQueryRecordConverter(SHOULD_CONVERT_DOUBLE)
+        .convertRecord(kafkaConnectRecord);
+    Assert.assertEquals(kafkaConnectMap, stringObjectMap);
+  }
+
   @Test
   public void testMapSchemalessConvertDouble() {
     Map kafkaConnectMap = new HashMap<Object, Object>(){{