Repository: microsoft/kafka-connect-cosmosdb
Pull request #441 (2022-02-01 21:57:22)

Release 1.2.5 version update
* Release 1.2.5 version update and changelog.
* Also includes nullable struct fix 



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 846107e..787c265 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,9 @@
 ## Release History
 
+### 1.2.5 (2022-02-01)
+#### Key Bug Fixes
+* Fixed NullPointerException issue with nullable structs. See this [PR 439](https://github.com/microsoft/kafka-connect-cosmosdb/pull/439).
+
 ### 1.2.4 (2022-01-05)
 #### Key Bug Fixes
 * Upgraded log4j2 to 2.17.1 to address the security vulnerability
diff --git a/pom.xml b/pom.xml
index eb0ed2d..c8f271f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -7,7 +7,7 @@
 
     <groupId>com.azure.cosmos.kafka</groupId>
     <artifactId>kafka-connect-cosmos</artifactId>
-    <version>1.2.4</version>
+    <version>1.2.5</version>
 
     <name> kafka-connect-cosmos</name>
     <url>https://github.com/microsoft/kafka-connect-cosmosdb</url>
diff --git a/src/main/java/com/azure/cosmos/kafka/connect/sink/StructToJsonMap.java b/src/main/java/com/azure/cosmos/kafka/connect/sink/StructToJsonMap.java
index fccea3c..207b89e 100644
--- a/src/main/java/com/azure/cosmos/kafka/connect/sink/StructToJsonMap.java
+++ b/src/main/java/com/azure/cosmos/kafka/connect/sink/StructToJsonMap.java
@@ -15,6 +15,9 @@ import org.apache.kafka.connect.data.Timestamp;
 public class StructToJsonMap {
 
     public static Map<String, Object> toJsonMap(Struct struct) {
+        if (struct == null) {
+            return null;
+        }
         Map<String, Object> jsonMap = new HashMap<String, Object>(0);
         List<Field> fields = struct.schema().fields();
         for (Field field : fields) {


Test patch:
diff --git a/src/test/java/com/azure/cosmos/kafka/connect/sink/StructToJsonMapTest.java b/src/test/java/com/azure/cosmos/kafka/connect/sink/StructToJsonMapTest.java
index c66b906..009971a 100644
--- a/src/test/java/com/azure/cosmos/kafka/connect/sink/StructToJsonMapTest.java
+++ b/src/test/java/com/azure/cosmos/kafka/connect/sink/StructToJsonMapTest.java
@@ -19,59 +19,60 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
 
 public class StructToJsonMapTest {
+
     @Test
     public void emptyStructToEmptyMap() {
         Schema schema = SchemaBuilder.struct()
-            .build();
+                .build();
         Struct struct = new Struct(schema);
         assertEquals(ImmutableMap.of(), StructToJsonMap.toJsonMap(struct));
     }
 
     @Test
-    public void complextStructToMap() {
+    public void complexStructToMap() {
         Schema embeddedSchema = SchemaBuilder.struct()
-            .field("embedded_string", Schema.STRING_SCHEMA)
-            .build();
+                .field("embedded_string", Schema.STRING_SCHEMA)
+                .build();
         Schema schema = SchemaBuilder.struct()
-            .field("boolean", Schema.BOOLEAN_SCHEMA)
-            .field("int8", Schema.INT8_SCHEMA)
-            .field("int16", Schema.INT16_SCHEMA)
-            .field("int32", Schema.INT32_SCHEMA)
-            .field("int64", Schema.INT64_SCHEMA)
-            .field("float32", Schema.FLOAT32_SCHEMA)
-            .field("float64", Schema.FLOAT64_SCHEMA)
-            .field("date", Date.SCHEMA)
-            .field("time", Time.SCHEMA)
-            .field("timestamp", Timestamp.SCHEMA)
-            .field("decimal", Decimal.schema(10))
-            .field("string", Schema.STRING_SCHEMA)
-            .field("struct", embeddedSchema)
-            .field("array_of_boolean", SchemaBuilder.array(Schema.BOOLEAN_SCHEMA).build())
-            .field("array_of_struct", SchemaBuilder.array(embeddedSchema).build())
-            .field("optional_string", Schema.OPTIONAL_STRING_SCHEMA)
-            .build();
+                .field("boolean", Schema.BOOLEAN_SCHEMA)
+                .field("int8", Schema.INT8_SCHEMA)
+                .field("int16", Schema.INT16_SCHEMA)
+                .field("int32", Schema.INT32_SCHEMA)
+                .field("int64", Schema.INT64_SCHEMA)
+                .field("float32", Schema.FLOAT32_SCHEMA)
+                .field("float64", Schema.FLOAT64_SCHEMA)
+                .field("date", Date.SCHEMA)
+                .field("time", Time.SCHEMA)
+                .field("timestamp", Timestamp.SCHEMA)
+                .field("decimal", Decimal.schema(10))
+                .field("string", Schema.STRING_SCHEMA)
+                .field("struct", embeddedSchema)
+                .field("array_of_boolean", SchemaBuilder.array(Schema.BOOLEAN_SCHEMA).build())
+                .field("array_of_struct", SchemaBuilder.array(embeddedSchema).build())
+                .field("optional_string", Schema.OPTIONAL_STRING_SCHEMA)
+                .build();
 
         java.util.Date now = new java.util.Date();
         BigDecimal ten = new BigDecimal(10).setScale(10);
         String quickBrownFox = "The quick brown fox jumps over the lazy dog";
         Struct struct = new Struct(schema)
-            .put("boolean", false)
-            .put("int8", (byte) 0)
-            .put("int16", (short) 1)
-            .put("int32", 2)
-            .put("int64", (long) 3)
-            .put("float32", (float) 4.0)
-            .put("float64", (double) 5.0)
-            .put("date", now)
-            .put("time", now)
-            .put("timestamp", now)
-            .put("decimal", ten)
-            .put("string", quickBrownFox)
-            .put("struct", new Struct(embeddedSchema)
-                .put("embedded_string", quickBrownFox))
-            .put("array_of_boolean", ImmutableList.of(false))
-            .put("array_of_struct", ImmutableList.of(
-                new Struct(embeddedSchema).put("embedded_string", quickBrownFox)));
+                .put("boolean", false)
+                .put("int8", (byte) 0)
+                .put("int16", (short) 1)
+                .put("int32", 2)
+                .put("int64", (long) 3)
+                .put("float32", (float) 4.0)
+                .put("float64", (double) 5.0)
+                .put("date", now)
+                .put("time", now)
+                .put("timestamp", now)
+                .put("decimal", ten)
+                .put("string", quickBrownFox)
+                .put("struct", new Struct(embeddedSchema)
+                        .put("embedded_string", quickBrownFox))
+                .put("array_of_boolean", ImmutableList.of(false))
+                .put("array_of_struct", ImmutableList.of(
+                        new Struct(embeddedSchema).put("embedded_string", quickBrownFox)));
 
         Map<String, Object> converted = StructToJsonMap.toJsonMap(struct);
         assertEquals(false, converted.get("boolean"));
@@ -91,4 +92,26 @@ public class StructToJsonMapTest {
         assertEquals(ImmutableMap.of("embedded_string", quickBrownFox), ((List<Struct>) converted.get("array_of_struct")).get(0));
         assertNull(converted.get("optional_string"));
     }
+
+    @Test
+    public void nullableStructTest() {
+        Schema embeddedSchema = SchemaBuilder.struct()
+                .optional()
+                .field("embedded_string", Schema.STRING_SCHEMA)
+                .field("embedded_int", Schema.INT32_SCHEMA)
+                .build();
+        Schema schema = SchemaBuilder.struct()
+                .field("struct", embeddedSchema)
+                .field("string", Schema.STRING_SCHEMA)
+                .build();
+
+        Struct struct = new Struct(schema)
+                .put("struct", null)
+                .put("string", "string_value");
+
+        Map<String, Object> converted = StructToJsonMap.toJsonMap(struct);
+        assertEquals("string_value", converted.get("string"));
+        assertNull(converted.get("struct"));
+    }
+
 }