From a946446754849d62eb384a8cfed509a3de381022 Mon Sep 17 00:00:00 2001
From: Oryan Moshe <oryan@moshe.ai>
Date: Tue, 21 Apr 2020 18:02:47 +0300
Subject: [PATCH] Added handeling for null values

---
 pom.xml                                                      | 2 +-
 .../oryanmoshe/kafka/connect/util/TimestampConverter.java    | 5 ++++-
 2 files changed, 5 insertions(+), 2 deletions(-)

diff --git a/pom.xml b/pom.xml
index afb63ad..3fe8e08 100644
--- a/pom.xml
+++ b/pom.xml
@@ -3,7 +3,7 @@
   <groupId>oryanmoshe.kafka.connect.util</groupId>
   <artifactId>TimestampConverter</artifactId>
   <packaging>jar</packaging>
-  <version>1.1.1-SNAPSHOT</version>
+  <version>1.1.2-SNAPSHOT</version>
   <name>TimestampConverter</name>
   <url>http://maven.apache.org</url>
 
diff --git a/src/main/java/oryanmoshe/kafka/connect/util/TimestampConverter.java b/src/main/java/oryanmoshe/kafka/connect/util/TimestampConverter.java
index 58a6fda..6c14c43 100644
--- a/src/main/java/oryanmoshe/kafka/connect/util/TimestampConverter.java
+++ b/src/main/java/oryanmoshe/kafka/connect/util/TimestampConverter.java
@@ -71,6 +71,9 @@ public void converterFor(RelationalColumn column, ConverterRegistration<SchemaBu
         if (SUPPORTED_DATA_TYPES.stream().anyMatch(s -> s.toLowerCase().equals(column.typeName().toLowerCase()))) {
             boolean isTime = "time".equals(column.typeName().toLowerCase());
             registration.register(datetimeSchema, rawValue -> {
+                if (rawValue == null)
+                    return rawValue;
+
                 Long millis = getMillis(rawValue.toString(), isTime);
                 if (millis == null)
                     return rawValue.toString();
@@ -89,7 +92,7 @@ public void converterFor(RelationalColumn column, ConverterRegistration<SchemaBu
     }
 
     private Long getMillis(String timestamp, boolean isTime) {
-        if (timestamp == null || timestamp.isBlank())
+        if (timestamp.isBlank())
             return null;
 
         if (timestamp.contains(":") || timestamp.contains("-")) {