Merge pull request #94 from data-integrations/feature/CDAP-17386-handle-null-value-0.1

Handle null value correctly to avoid NPE

Author: seanzhougoogle

delta-plugins-common/src/main/java/io/cdap/delta/plugin/common/Records.java

@@ -161,7 +161,7 @@ public static StructuredRecord convert(Struct struct) {
       Schema fieldSchema = field.getSchema();
       fieldSchema = fieldSchema.isNullable() ? fieldSchema.getNonNullable() : fieldSchema;
       Schema.LogicalType logicalType = fieldSchema.getLogicalType();
-      if (logicalType == null) {
+      if (logicalType == null || val == null) {
         builder.set(fieldName, val);
       } else {
         switch (logicalType) {

mysql-delta-plugins/src/test/java/io/cdap/delta/mysql/MySqlEventReaderIntegrationTest.java

@@ -122,6 +122,11 @@ public static void setupClass() throws Exception {
         ps.setDate(3, Date.valueOf("1971-01-01"));
         ps.addBatch();
 
+        ps.setInt(1, 2);
+        ps.setString(2, "tim");
+        ps.setDate(3, null);
+        ps.addBatch();
+
         ps.executeBatch();
       }
     }
@@ -190,6 +195,18 @@ public void test() throws InterruptedException {
       .setDate("bday", LocalDate.ofEpochDay(365))
       .build();
     Assert.assertEquals(expected, row);
+
+    dmlEvent = eventEmitter.getDmlEvents().get(2);
+    Assert.assertEquals(DMLOperation.INSERT, dmlEvent.getOperation());
+    Assert.assertEquals(DB, dmlEvent.getDatabase());
+    Assert.assertEquals(CUSTOMERS_TABLE, dmlEvent.getTable());
+    row = dmlEvent.getRow();
+    expected = StructuredRecord.builder(CUSTOMERS_SCHEMA)
+      .set("id", 2)
+      .set("name", "tim")
+      .setDate("bday", null)
+      .build();
+    Assert.assertEquals(expected, row);
   }
 
   @Test

sqlserver-delta-plugins/src/test/java/io.cdap.delta.sqlserver/SqlServerEventReaderIntegrationTest.java

@@ -119,6 +119,11 @@ public static void setupClass() throws Exception {
         ps.setDate(3, Date.valueOf("1971-01-01"));
         ps.addBatch();
 
+        ps.setInt(1, 2);
+        ps.setString(2, "tim");
+        ps.setDate(3, null);
+        ps.addBatch();
+
         ps.executeBatch();
       }
 
@@ -194,6 +199,18 @@ public void test() throws InterruptedException {
       .setDate("bday", LocalDate.ofEpochDay(365))
       .build();
     Assert.assertEquals(expected, row);
+
+    dmlEvent = eventEmitter.getDmlEvents().get(2);
+    Assert.assertEquals(DMLOperation.INSERT, dmlEvent.getOperation());
+    Assert.assertEquals(DB, dmlEvent.getDatabase());
+    Assert.assertEquals(CUSTOMERS_TABLE, dmlEvent.getTable());
+    row = dmlEvent.getRow();
+    expected = StructuredRecord.builder(CUSTOMERS_SCHEMA)
+      .set("id", 2)
+      .set("name", "tim")
+      .setDate("bday", null)
+      .build();
+    Assert.assertEquals(expected, row);
   }
 
   @Test