KAFKA-4100: Ensure 'fields' and 'fieldsByName' are not null for Struct schemas

Author: Shikhar Bhushan <[email protected]>

Reviewers: Ewen Cheslack-Postava <[email protected]>

Closes apache#1800 from shikhar/kafka-4100

Author: shikhar

connect/api/src/main/java/org/apache/kafka/connect/data/ConnectSchema.java

@@ -22,6 +22,7 @@
 import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -106,12 +107,13 @@ public ConnectSchema(Type type, boolean optional, Object defaultValue, String na
         this.doc = doc;
         this.parameters = parameters;
 
-        this.fields = fields;
-        if (this.fields != null && this.type == Type.STRUCT) {
-            this.fieldsByName = new HashMap<>();
-            for (Field field : fields)
+        if (this.type == Type.STRUCT) {
+            this.fields = fields == null ? Collections.<Field>emptyList() : fields;
+            this.fieldsByName = new HashMap<>(this.fields.size());
+            for (Field field : this.fields)
                 fieldsByName.put(field.name(), field);
         } else {
+            this.fields = null;
             this.fieldsByName = null;
         }
 

connect/api/src/main/java/org/apache/kafka/connect/data/SchemaBuilder.java

@@ -78,6 +78,9 @@ public class SchemaBuilder implements Schema {
 
     private SchemaBuilder(Type type) {
         this.type = type;
+        if (type == Type.STRUCT) {
+            fields = new ArrayList<>();
+        }
     }
 
     // Common/metadata fields
@@ -317,8 +320,6 @@ public static SchemaBuilder struct() {
     public SchemaBuilder field(String fieldName, Schema fieldSchema) {
         if (type != Type.STRUCT)
             throw new SchemaBuilderException("Cannot create fields on type " + type);
-        if (fields == null)
-            fields = new ArrayList<>();
         int fieldIndex = fields.size();
         fields.add(new Field(fieldName, fieldIndex, fieldSchema));
         return this;

connect/api/src/test/java/org/apache/kafka/connect/data/ConnectSchemaTest.java

@@ -300,4 +300,11 @@ public void testStructEquality() {
         assertNotEquals(s1, differentField);
     }
 
+    @Test
+    public void testEmptyStruct() {
+        final ConnectSchema emptyStruct = new ConnectSchema(Schema.Type.STRUCT, false, null, null, null, null);
+        assertEquals(0, emptyStruct.fields().size());
+        new Struct(emptyStruct);
+    }
+
 }

connect/api/src/test/java/org/apache/kafka/connect/data/SchemaBuilderTest.java

@@ -283,7 +283,16 @@ public void testMapBuilderInvalidDefault() {
                 .defaultValue(defMap).build();
     }
 
-
+    @Test
+    public void testEmptyStruct() {
+        final SchemaBuilder emptyStructSchemaBuilder = SchemaBuilder.struct();
+        assertEquals(0, emptyStructSchemaBuilder.fields().size());
+        new Struct(emptyStructSchemaBuilder);
+
+        final Schema emptyStructSchema = emptyStructSchemaBuilder.build();
+        assertEquals(0, emptyStructSchema.fields().size());
+        new Struct(emptyStructSchema);
+    }
 
     private void assertTypeAndDefault(Schema schema, Schema.Type type, boolean optional, Object defaultValue) {
         assertEquals(type, schema.type());