KAFKA-3690: Avoid to pass null to UnmodifiableMap

Author: Liquan Pei <[email protected]> Reviewers: Jason Gustafson <[email protected]>, Ismael Juma <[email protected]>, Ewen Cheslack-Postava <[email protected]> Closes apache#1360 from Ishiihara/avoid-to-pass-null
priyank5485 · May 11, 2016 · bd8681c · bd8681c
1 parent 6978115
commit bd8681c
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 3 deletions.
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
@@ -181,7 +181,7 @@ public SchemaBuilder doc(String doc) {
 
     @Override
     public Map<String, String> parameters() {
-        return Collections.unmodifiableMap(parameters);
+        return parameters == null ? null : Collections.unmodifiableMap(parameters);
     }
 
     /**

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
@@ -252,7 +252,14 @@ public void testArrayBuilderInvalidDefault() {
 
     @Test
     public void testMapBuilder() {
-        Schema schema = SchemaBuilder.map(Schema.INT8_SCHEMA, Schema.INT8_SCHEMA).build();
+        // SchemaBuilder should also pass the check
+        Schema schema = SchemaBuilder.map(Schema.INT8_SCHEMA, Schema.INT8_SCHEMA);
+        assertTypeAndDefault(schema, Schema.Type.MAP, false, null);
+        assertEquals(schema.keySchema(), Schema.INT8_SCHEMA);
+        assertEquals(schema.valueSchema(), Schema.INT8_SCHEMA);
+        assertNoMetadata(schema);
+
+        schema = SchemaBuilder.map(Schema.INT8_SCHEMA, Schema.INT8_SCHEMA).build();
         assertTypeAndDefault(schema, Schema.Type.MAP, false, null);
         assertEquals(schema.keySchema(), Schema.INT8_SCHEMA);
         assertEquals(schema.valueSchema(), Schema.INT8_SCHEMA);

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java
@@ -179,7 +179,7 @@ public static String COMMIT_TASKS_KEY(String connectorName) {
     // converter/serializer changes causing keys to change. We need to absolutely ensure that the keys remain precisely
     // the same.
     public static final Schema CONNECTOR_CONFIGURATION_V0 = SchemaBuilder.struct()
-            .field("properties", SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.OPTIONAL_STRING_SCHEMA))
+            .field("properties", SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.OPTIONAL_STRING_SCHEMA).build())
             .build();
     public static final Schema TASK_CONFIGURATION_V0 = CONNECTOR_CONFIGURATION_V0;
     public static final Schema CONNECTOR_TASKS_COMMIT_V0 = SchemaBuilder.struct()