Repository: kafka Updated Branches: refs/heads/trunk 9aaeb33c1 -> b91eeac94
KAFKA-4100: Ensure 'fields' and 'fieldsByName' are not null for Struct schemas Author: Shikhar Bhushan <[email protected]> Reviewers: Ewen Cheslack-Postava <[email protected]> Closes #1800 from shikhar/kafka-4100 Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/b91eeac9 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/b91eeac9 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/b91eeac9 Branch: refs/heads/trunk Commit: b91eeac9438b8718c410045b0e9191296ebb536d Parents: 9aaeb33 Author: Shikhar Bhushan <[email protected]> Authored: Mon Aug 29 19:08:52 2016 -0700 Committer: Ewen Cheslack-Postava <[email protected]> Committed: Mon Aug 29 19:08:52 2016 -0700 ---------------------------------------------------------------------- .../org/apache/kafka/connect/data/ConnectSchema.java | 10 ++++++---- .../org/apache/kafka/connect/data/SchemaBuilder.java | 5 +++-- .../org/apache/kafka/connect/data/ConnectSchemaTest.java | 7 +++++++ .../org/apache/kafka/connect/data/SchemaBuilderTest.java | 11 ++++++++++- 4 files changed, 26 insertions(+), 7 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/b91eeac9/connect/api/src/main/java/org/apache/kafka/connect/data/ConnectSchema.java ---------------------------------------------------------------------- diff --git a/connect/api/src/main/java/org/apache/kafka/connect/data/ConnectSchema.java b/connect/api/src/main/java/org/apache/kafka/connect/data/ConnectSchema.java index 591644c..d1fd9cd 100644 --- a/connect/api/src/main/java/org/apache/kafka/connect/data/ConnectSchema.java +++ b/connect/api/src/main/java/org/apache/kafka/connect/data/ConnectSchema.java @@ -22,6 +22,7 @@ import org.apache.kafka.connect.errors.DataException; 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 class ConnectSchema implements Schema { 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; } http://git-wip-us.apache.org/repos/asf/kafka/blob/b91eeac9/connect/api/src/main/java/org/apache/kafka/connect/data/SchemaBuilder.java ---------------------------------------------------------------------- diff --git a/connect/api/src/main/java/org/apache/kafka/connect/data/SchemaBuilder.java b/connect/api/src/main/java/org/apache/kafka/connect/data/SchemaBuilder.java index 32045f9..2ec1cf3 100644 --- a/connect/api/src/main/java/org/apache/kafka/connect/data/SchemaBuilder.java +++ b/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 class SchemaBuilder implements Schema { 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; http://git-wip-us.apache.org/repos/asf/kafka/blob/b91eeac9/connect/api/src/test/java/org/apache/kafka/connect/data/ConnectSchemaTest.java ---------------------------------------------------------------------- diff --git a/connect/api/src/test/java/org/apache/kafka/connect/data/ConnectSchemaTest.java b/connect/api/src/test/java/org/apache/kafka/connect/data/ConnectSchemaTest.java index 4388ade..f5c6e2f 100644 --- a/connect/api/src/test/java/org/apache/kafka/connect/data/ConnectSchemaTest.java +++ b/connect/api/src/test/java/org/apache/kafka/connect/data/ConnectSchemaTest.java @@ -300,4 +300,11 @@ public class ConnectSchemaTest { 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); + } + } http://git-wip-us.apache.org/repos/asf/kafka/blob/b91eeac9/connect/api/src/test/java/org/apache/kafka/connect/data/SchemaBuilderTest.java ---------------------------------------------------------------------- diff --git a/connect/api/src/test/java/org/apache/kafka/connect/data/SchemaBuilderTest.java b/connect/api/src/test/java/org/apache/kafka/connect/data/SchemaBuilderTest.java index fdbaa0a..37cc47c 100644 --- a/connect/api/src/test/java/org/apache/kafka/connect/data/SchemaBuilderTest.java +++ b/connect/api/src/test/java/org/apache/kafka/connect/data/SchemaBuilderTest.java @@ -283,7 +283,16 @@ public class SchemaBuilderTest { .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());
