Add support for handling empty schemas and records (#237)

Sometimes record types contain fields of type that do not have any fields defined (null objects, marker classes etc.).
Converters must omit them since it's impossible to map empty structures into BigQuery table schema and rows.

Author: mkubala

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

@@ -178,12 +178,17 @@ private Map<String, Object> convertStruct(Object kafkaConnectObject, Schema kafk
     List<Field> kafkaConnectSchemaFields = kafkaConnectSchema.fields();
     Struct kafkaConnectStruct = (Struct) kafkaConnectObject;
     for (Field kafkaConnectField : kafkaConnectSchemaFields) {
-      Object bigQueryObject = convertObject(
-          kafkaConnectStruct.get(kafkaConnectField.name()),
-          kafkaConnectField.schema()
-      );
-      if (bigQueryObject != null) {
-        bigQueryRecord.put(kafkaConnectField.name(), bigQueryObject);
+      // ignore empty structures
+      boolean isEmptyStruct = kafkaConnectField.schema().type() == Schema.Type.STRUCT &&
+          kafkaConnectField.schema().fields().isEmpty();
+      if (!isEmptyStruct) {
+        Object bigQueryObject = convertObject(
+            kafkaConnectStruct.get(kafkaConnectField.name()),
+            kafkaConnectField.schema()
+        );
+        if (bigQueryObject != null) {
+          bigQueryRecord.put(kafkaConnectField.name(), bigQueryObject);
+        }
       }
     }
     return bigQueryRecord;

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

@@ -27,14 +27,14 @@
 import com.wepay.kafka.connect.bigquery.convert.logicaltype.LogicalTypeConverter;
 import com.wepay.kafka.connect.bigquery.exception.ConversionConnectException;
 
-import org.apache.kafka.connect.data.Field;
 import org.apache.kafka.connect.data.Schema;
 
-import java.util.ArrayList;
 import java.util.HashMap;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 /**
  * Class for converting from {@link Schema Kafka Connect Schemas} to
@@ -102,25 +102,26 @@ public com.google.cloud.bigquery.Schema convertSchema(Schema kafkaConnectSchema)
           ConversionConnectException("Top-level Kafka Connect schema must be of type 'struct'");
     }
 
-    List<com.google.cloud.bigquery.Field> fields = new LinkedList<>();
-
-    for (Field kafkaConnectField : kafkaConnectSchema.fields()) {
-      com.google.cloud.bigquery.Field bigQuerySchemaField =
-          convertField(kafkaConnectField.schema(), kafkaConnectField.name()).build();
-      fields.add(bigQuerySchemaField);
-    }
+    List<com.google.cloud.bigquery.Field> fields = kafkaConnectSchema.fields().stream()
+        .flatMap(kafkaConnectField ->
+          convertField(kafkaConnectField.schema(), kafkaConnectField.name())
+          .map(Stream::of)
+          .orElse(Stream.empty())
+        )
+        .map(com.google.cloud.bigquery.Field.Builder::build)
+        .collect(Collectors.toList());
 
     return com.google.cloud.bigquery.Schema.of(fields);
   }
 
-  private com.google.cloud.bigquery.Field.Builder convertField(Schema kafkaConnectSchema,
-                                                               String fieldName) {
-    com.google.cloud.bigquery.Field.Builder result;
+  private Optional<com.google.cloud.bigquery.Field.Builder> convertField(Schema kafkaConnectSchema,
+                                                                         String fieldName) {
+    Optional<com.google.cloud.bigquery.Field.Builder> result;
     Schema.Type kafkaConnectSchemaType = kafkaConnectSchema.type();
     if (LogicalConverterRegistry.isRegisteredLogicalType(kafkaConnectSchema.name())) {
-      result = convertLogical(kafkaConnectSchema, fieldName);
+      result = Optional.of(convertLogical(kafkaConnectSchema, fieldName));
     } else if (PRIMITIVE_TYPE_MAP.containsKey(kafkaConnectSchemaType)) {
-      result = convertPrimitive(kafkaConnectSchema, fieldName);
+      result = Optional.of(convertPrimitive(kafkaConnectSchema, fieldName));
     } else {
       switch (kafkaConnectSchemaType) {
         case STRUCT:
@@ -138,11 +139,13 @@ private com.google.cloud.bigquery.Field.Builder convertField(Schema kafkaConnect
           );
       }
     }
-    setNullability(kafkaConnectSchema, result);
-    if (kafkaConnectSchema.doc() != null) {
-      result.setDescription(kafkaConnectSchema.doc());
-    }
-    return result;
+    return result.map(res -> {
+      setNullability(kafkaConnectSchema, res);
+      if (kafkaConnectSchema.doc() != null) {
+        res.setDescription(kafkaConnectSchema.doc());
+      }
+      return res;
+    });
   }
 
   private void setNullability(Schema kafkaConnectSchema,
@@ -160,48 +163,54 @@ private void setNullability(Schema kafkaConnectSchema,
     }
   }
 
-  private com.google.cloud.bigquery.Field.Builder convertStruct(Schema kafkaConnectSchema,
-                                                                String fieldName) {
-    List<com.google.cloud.bigquery.Field> bigQueryRecordFields = new ArrayList<>();
-
-    for (Field kafkaConnectField : kafkaConnectSchema.fields()) {
-      com.google.cloud.bigquery.Field.Builder bigQueryRecordFieldBuilder =
-          convertField(kafkaConnectField.schema(), kafkaConnectField.name());
-      bigQueryRecordFields.add(bigQueryRecordFieldBuilder.build());
+  private Optional<com.google.cloud.bigquery.Field.Builder> convertStruct(Schema kafkaConnectSchema,
+                                                                          String fieldName) {
+    List<com.google.cloud.bigquery.Field> bigQueryRecordFields = kafkaConnectSchema.fields()
+        .stream()
+        .flatMap(kafkaConnectField ->
+            convertField(kafkaConnectField.schema(), kafkaConnectField.name())
+            .map(Stream::of)
+            .orElse(Stream.empty())
+        )
+        .map(com.google.cloud.bigquery.Field.Builder::build)
+        .collect(Collectors.toList());
+    if (bigQueryRecordFields.isEmpty()) {
+      return Optional.empty();
     }
 
     FieldList fieldList = FieldList.of(bigQueryRecordFields);
 
-    return com.google.cloud.bigquery.Field.newBuilder(fieldName,
-                                                      LegacySQLTypeName.RECORD,
-                                                      fieldList);
+    return Optional.of(com.google.cloud.bigquery.Field.newBuilder(fieldName,
+                                                                  LegacySQLTypeName.RECORD,
+                                                                  fieldList));
   }
 
-  private com.google.cloud.bigquery.Field.Builder convertArray(Schema kafkaConnectSchema,
-                                                               String fieldName) {
+  private Optional<com.google.cloud.bigquery.Field.Builder> convertArray(Schema kafkaConnectSchema,
+                                                                         String fieldName) {
     Schema elementSchema = kafkaConnectSchema.valueSchema();
-    com.google.cloud.bigquery.Field.Builder elementFieldBuilder =
-        convertField(elementSchema, fieldName);
-    return elementFieldBuilder.setMode(com.google.cloud.bigquery.Field.Mode.REPEATED);
+    return convertField(elementSchema, fieldName)
+        .map(builder -> builder.setMode(com.google.cloud.bigquery.Field.Mode.REPEATED));
   }
 
-  private com.google.cloud.bigquery.Field.Builder convertMap(Schema kafkaConnectSchema,
-                                                             String fieldName) {
+  private Optional<com.google.cloud.bigquery.Field.Builder> convertMap(Schema kafkaConnectSchema,
+                                                                       String fieldName) {
     Schema keySchema = kafkaConnectSchema.keySchema();
     Schema valueSchema = kafkaConnectSchema.valueSchema();
 
-    com.google.cloud.bigquery.Field keyField =
-        convertField(keySchema, MAP_KEY_FIELD_NAME).build();
-    com.google.cloud.bigquery.Field valueField =
-        convertField(valueSchema, MAP_VALUE_FIELD_NAME).build();
-
-    com.google.cloud.bigquery.Field.Builder bigQueryRecordBuilder =
-        com.google.cloud.bigquery.Field.newBuilder(fieldName,
-                                                   LegacySQLTypeName.RECORD,
-                                                   keyField,
-                                                   valueField);
-
-    return bigQueryRecordBuilder.setMode(com.google.cloud.bigquery.Field.Mode.REPEATED);
+    Optional<com.google.cloud.bigquery.Field> maybeKeyField = convertField(keySchema, MAP_KEY_FIELD_NAME)
+        .map(com.google.cloud.bigquery.Field.Builder::build);
+    Optional<com.google.cloud.bigquery.Field> maybeValueField = convertField(valueSchema, MAP_VALUE_FIELD_NAME)
+        .map(com.google.cloud.bigquery.Field.Builder::build);
+
+    return maybeKeyField.flatMap(keyField ->
+        maybeValueField.map(valueField ->
+            com.google.cloud.bigquery.Field.newBuilder(fieldName,
+                                                       LegacySQLTypeName.RECORD,
+                                                       keyField,
+                                                       valueField)
+                .setMode(com.google.cloud.bigquery.Field.Mode.REPEATED)
+        )
+    );
   }
 
   private com.google.cloud.bigquery.Field.Builder convertPrimitive(Schema kafkaConnectSchema,

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

@@ -313,6 +313,56 @@ public void testStruct() {
     assertEquals(bigQueryExpectedOuterRecord, bigQueryTestOuterRecord);
   }
 
+  @Test
+  public void testEmptyStruct() {
+    Schema kafkaConnectInnerSchema = SchemaBuilder
+        .struct()
+        .build();
+
+    Struct kafkaConnectInnerStruct = new Struct(kafkaConnectInnerSchema);
+
+    SinkRecord kafkaConnectSinkRecord =
+        spoofSinkRecord(kafkaConnectInnerSchema, kafkaConnectInnerStruct, false);
+    Map<String, Object> bigQueryTestInnerRecord =
+        new BigQueryRecordConverter(SHOULD_CONVERT_DOUBLE)
+            .convertRecord(kafkaConnectSinkRecord, KafkaSchemaRecordType.VALUE);
+    assertEquals(new HashMap<String, Object>(), bigQueryTestInnerRecord);
+  }
+
+  @Test
+  public void testEmptyInnerStruct() {
+    final String innerFieldStructName = "InnerStruct";
+    final String innerFieldStringName = "InnerString";
+    final String innerStringValue = "forty two";
+
+    Schema kafkaConnectInnerSchema = SchemaBuilder
+        .struct()
+        .build();
+
+    Struct kafkaConnectInnerStruct = new Struct(kafkaConnectInnerSchema);
+
+    Schema kafkaConnectOuterSchema = SchemaBuilder
+        .struct()
+        .field(innerFieldStructName, kafkaConnectInnerSchema)
+        .field(innerFieldStringName, Schema.STRING_SCHEMA)
+        .build();
+
+    Struct kafkaConnectOuterStruct = new Struct(kafkaConnectOuterSchema);
+    kafkaConnectOuterStruct.put(innerFieldStructName, kafkaConnectInnerStruct);
+    kafkaConnectOuterStruct.put(innerFieldStringName, innerStringValue);
+
+    Map<String, Object> bigQueryExpectedOuterRecord = new HashMap<>();
+    bigQueryExpectedOuterRecord.put(innerFieldStringName, innerStringValue);
+
+    SinkRecord kafkaConnectOuterSinkRecord =
+        spoofSinkRecord(kafkaConnectOuterSchema, kafkaConnectOuterStruct, false);
+    Map<String, Object> bigQueryTestOuterRecord =
+        new BigQueryRecordConverter(SHOULD_CONVERT_DOUBLE)
+            .convertRecord(kafkaConnectOuterSinkRecord, KafkaSchemaRecordType.VALUE);
+
+    assertEquals(bigQueryExpectedOuterRecord, bigQueryTestOuterRecord);
+  }
+
   @Test
   public void testMap() {
     final String fieldName = "StringIntegerMap";

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

@@ -275,6 +275,61 @@ public void testStruct() { // Struct in a struct in a struct (wrapped in a struc
     assertEquals(bigQueryExpectedOuterSchema, bigQueryTestOuterSchema);
   }
 
+  @Test
+  public void testEmptyStruct() { // Empty struct
+    com.google.cloud.bigquery.Schema bigQueryTestOuterSchema =
+        new BigQuerySchemaConverter(false).convertSchema(
+            SchemaBuilder
+                .struct()
+                .build()
+        );
+    assertEquals(com.google.cloud.bigquery.Schema.of(), bigQueryTestOuterSchema);
+  }
+
+  @Test
+  public void testEmptyInnerStruct() { // Empty nested struct
+    final String outerFieldStructName = "OuterStruct";
+    final String innerFieldStructName = "InnerStruct";
+    final String innerFieldStringName = "InnerString";
+
+    Schema kafkaConnectInnerSchema = SchemaBuilder
+        .struct()
+        .build();
+
+    com.google.cloud.bigquery.Field bigQueryInnerString = com.google.cloud.bigquery.Field.newBuilder(
+        innerFieldStringName,
+        LegacySQLTypeName.STRING
+    ).setMode(
+        com.google.cloud.bigquery.Field.Mode.REQUIRED
+    ).build();
+
+    com.google.cloud.bigquery.Field bigQueryOuterRecord =
+        com.google.cloud.bigquery.Field.newBuilder(
+            outerFieldStructName,
+            LegacySQLTypeName.RECORD,
+            bigQueryInnerString
+        ).setMode(
+            com.google.cloud.bigquery.Field.Mode.REQUIRED
+        ).build();
+
+    Schema kafkaConnectOuterSchema = SchemaBuilder
+        .struct()
+        .field(innerFieldStructName, kafkaConnectInnerSchema)
+        .field(innerFieldStringName, Schema.STRING_SCHEMA)
+        .build();
+
+    com.google.cloud.bigquery.Schema bigQueryExpectedOuterSchema =
+        com.google.cloud.bigquery.Schema.of(bigQueryOuterRecord);
+    com.google.cloud.bigquery.Schema bigQueryTestOuterSchema =
+        new BigQuerySchemaConverter(false).convertSchema(
+            SchemaBuilder
+                .struct()
+                .field(outerFieldStructName, kafkaConnectOuterSchema)
+                .build()
+        );
+    assertEquals(bigQueryExpectedOuterSchema, bigQueryTestOuterSchema);
+  }
+
   @Test
   public void testMap() {
     final String fieldName = "StringIntegerMap";