Added null check. (#343)

Added null check for handling TimestampLTZ type.

Author: MrRahulSharma

oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceDBRecord.java

@@ -35,6 +35,7 @@
 import java.sql.ResultSet;
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
+import java.sql.Timestamp;
 import java.sql.Types;
 import java.time.Instant;
 import java.time.ZoneId;
@@ -191,8 +192,9 @@ private void handleOracleSpecificType(ResultSet resultSet, StructuredRecord.Buil
         recordBuilder.set(field.getName(), resultSet.getString(columnIndex));
         break;
       case OracleSourceSchemaReader.TIMESTAMP_LTZ:
-        Instant instant = resultSet.getTimestamp(columnIndex).toInstant();
-        recordBuilder.setTimestamp(field.getName(), instant.atZone(ZoneId.ofOffset("UTC", ZoneOffset.UTC)));
+        Timestamp timestamp = resultSet.getTimestamp(columnIndex);
+        recordBuilder.setTimestamp(field.getName(), (timestamp != null) ?
+                timestamp.toInstant().atZone(ZoneId.ofOffset("UTC", ZoneOffset.UTC)) : null);
         break;
       case OracleSourceSchemaReader.BINARY_FLOAT:
         recordBuilder.set(field.getName(), resultSet.getFloat(columnIndex));

oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleSourceDBRecordUnitTest.java

@@ -143,4 +143,27 @@ public void validateDefaultDecimalParsing() throws Exception {
     Assert.assertTrue(record.getDecimal("ID4") instanceof BigDecimal);
     Assert.assertEquals(record.getDecimal("ID4").toPlainString(), "123.456789");
   }
+
+  /**
+   * Validate the Null value for TimestampLTZ datatype.
+   * @throws Exception
+   */
+  @Test
+  public void validateTimestampLTZTypeNullHandling() throws Exception {
+    Schema.Field field1 = Schema.Field.of("field1", Schema.nullableOf(Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)));
+
+    Schema schema = Schema.recordOf(
+            "dbRecord",
+            field1
+    );
+
+    when(resultSet.getTimestamp(eq(1))).thenReturn(null);
+
+    StructuredRecord.Builder builder = StructuredRecord.builder(schema);
+    OracleSourceDBRecord dbRecord = new OracleSourceDBRecord(null, null);
+    dbRecord.handleField(resultSet, builder, field1, 1, OracleSourceSchemaReader.TIMESTAMP_LTZ, DEFAULT_PRECISION, 0);
+
+    StructuredRecord record = builder.build();
+    Assert.assertNull(record.getTimestamp("field1"));
+  }
 }