Merge pull request #15 from oryanmoshe/fix/alonr/add-optional-column-…

…check-and-logging Use a new SchemaBuilder every time in order to avoid changing 'Already set' options
oryanmoshe · May 19, 2021 · e8dc33e · e8dc33e
2 parents ab2ab1b + 10a2847
commit e8dc33e
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 7 deletions.
diff --git a/src/main/java/oryanmoshe/kafka/connect/util/TimestampConverter.java b/src/main/java/oryanmoshe/kafka/connect/util/TimestampConverter.java
@@ -36,7 +36,7 @@ public class TimestampConverter implements CustomConverter<SchemaBuilder, Relati
     public String strDatetimeFormat, strDateFormat, strTimeFormat;
     public Boolean debug;
 
-    private final SchemaBuilder datetimeSchema = SchemaBuilder.string().optional().name("oryanmoshe.time.DateTimeString");
+//    private final SchemaBuilder datetimeSchema = SchemaBuilder.string().optional().name("oryanmoshe.time.DateTimeString");
 
     private SimpleDateFormat simpleDatetimeFormatter, simpleDateFormatter, simpleTimeFormatter;
 
@@ -70,8 +70,15 @@ public void converterFor(RelationalColumn column, ConverterRegistration<SchemaBu
                     column.name(), column.typeName(), column.hasDefaultValue(), column.defaultValue(), column.isOptional());
         if (SUPPORTED_DATA_TYPES.stream().anyMatch(s -> s.equalsIgnoreCase(column.typeName()))) {
             boolean isTime = "time".equalsIgnoreCase(column.typeName());
-            registration.register(datetimeSchema, rawValue -> {
+            // Use a new SchemaBuilder every time in order to avoid changing "Already set" options
+            // in the schema builder between tables.
+            registration.register(SchemaBuilder.string().optional(), rawValue -> {
                 if (rawValue == null) {
+                    // DEBUG
+                    if (this.debug) {
+                        System.out.printf("[TimestampConverter.converterFor] rawValue of %s is null.%n", column.name());
+                    }
+
                     if (column.isOptional()) {
                         return null;
                     }

diff --git a/src/test/java/oryanmoshe/kafka/connect/util/TimestampConverterTests.java b/src/test/java/oryanmoshe/kafka/connect/util/TimestampConverterTests.java
@@ -139,7 +139,7 @@ public String typeName() {
 
             @Override
             public String name() {
-                return null;
+                return "datecolumn";
             }
 
             @Override
@@ -199,7 +199,7 @@ public String typeName() {
 
             @Override
             public String name() {
-                return null;
+                return "timecolumn";
             }
 
             @Override
@@ -259,7 +259,7 @@ public String typeName() {
 
             @Override
             public String name() {
-                return null;
+                return "datetimecolumn";
             }
 
             @Override
@@ -319,7 +319,7 @@ public String typeName() {
 
             @Override
             public String name() {
-                return null;
+                return "datetime2column";
             }
 
             @Override
@@ -379,7 +379,7 @@ public String typeName() {
 
             @Override
             public String name() {
-                return null;
+                return "timestampcolumn";
             }
 
             @Override