Allow null values for source if schema allows it

pom.xml

@@ -22,7 +22,7 @@
   <name>Salesforce plugins</name>
   <groupId>io.cdap.plugin</groupId>
   <artifactId>salesforce-plugins</artifactId>
-  <version>1.3.0</version>
+  <version>1.3.1</version>
   <packaging>jar</packaging>
 
 
@@ -509,6 +509,7 @@
                 <!-- This file should be not be there, but I am excluding it for now -->
                 <exclude>*.rst</exclude>
                 <exclude>*.md</exclude>
+                <exclude>**/*.iml</exclude>
                 <exclude>**/*.cdap</exclude>
                 <exclude>**/*.yaml</exclude>
                 <exclude>**/*.md</exclude>

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/MapToRecordTransformer.java

@@ -46,7 +46,12 @@ private void transformRecord(Schema schema, Map<String, ?> record, StructuredRec
 
   private Object convertValue(String fieldName, Object value, Schema fieldSchema) {
     if (fieldSchema.isNullable()) {
-      return convertValue(fieldName, value, fieldSchema.getNonNullable());
+      return value == null ? null : convertValue(fieldName, value, fieldSchema.getNonNullable());
+    }
+
+    if (value == null) {
+      throw new RuntimeException(
+          String.format("Found null value for non nullable field %s", fieldName));
     }
 
     Schema.Type fieldSchemaType = fieldSchema.getType();

src/test/java/io/cdap/plugin/salesforce/plugin/source/batch/MapToRecordTransformerTest.java

@@ -26,6 +26,7 @@
 import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -90,4 +91,30 @@ public void testTransform() {
     Assert.assertEquals(ZonedDateTime.parse(arrayField.get(0).get("nested_timestamp"), DateTimeFormatter.ISO_DATE_TIME),
                         arrayRecord.get(0).getTimestamp("nested_timestamp", ZoneOffset.UTC));
   }
+
+  @Test
+  public void testNullableFields() {
+    Schema schema = Schema.recordOf("output",
+        Schema.Field.of("string_field", Schema.nullableOf(Schema.of(Schema.Type.STRING))),
+        Schema.Field.of("double_field", Schema.nullableOf(Schema.of(Schema.Type.DOUBLE))));
+    Map<String, Object> records = new HashMap<>();
+    records.put("string_field", null);
+    records.put("double_field", null);
+    MapToRecordTransformer recordTransformer = new MapToRecordTransformer();
+    StructuredRecord structuredRecord = recordTransformer.transform(schema, records);
+    Assert.assertNull(structuredRecord.get("string_field"));
+    Assert.assertNull(structuredRecord.get("double_field"));
+  }
+
+  @Test(expected = RuntimeException.class)
+  public void testNonNullableFieldWithNull() {
+    Schema schema = Schema.recordOf("output",
+        Schema.Field.of("string_field", Schema.nullableOf(Schema.of(Schema.Type.STRING))),
+        Schema.Field.of("double_field", Schema.of(Schema.Type.DOUBLE)));
+    Map<String, Object> records = new HashMap<>();
+    records.put("string_field", "");
+    records.put("double_field", null);
+    MapToRecordTransformer recordTransformer = new MapToRecordTransformer();
+    recordTransformer.transform(schema, records);
+  }
 }