support mysql timestamp data type correctly (#187)

support mysql timestamp data type correctly

Author: seanzhougoogle

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

@@ -30,6 +30,7 @@
 import io.debezium.time.NanoTimestamp;
 import io.debezium.time.Time;
 import io.debezium.time.Timestamp;
+import io.debezium.time.ZonedTimestamp;
 import org.apache.kafka.connect.data.Decimal;
 import org.apache.kafka.connect.data.Field;
 import org.apache.kafka.connect.data.SchemaBuilder;
@@ -204,7 +205,14 @@ public static StructuredRecord convert(Struct struct) {
             builder.setTimestamp(fieldName, getZonedDateTime((long) val, TimeUnit.MILLISECONDS));
             break;
           case TIMESTAMP_MICROS:
-            builder.setTimestamp(fieldName, getZonedDateTime((long) val, TimeUnit.MICROSECONDS));
+            if (val instanceof  Long) {
+              builder.setTimestamp(fieldName, getZonedDateTime((long) val, TimeUnit.MICROSECONDS));
+              break;
+            }
+            if (val instanceof String) {
+              builder.setTimestamp(fieldName, ZonedDateTime.parse((String)val));
+              break;
+            }
             break;
           case TIME_MILLIS:
             builder.setTime(fieldName, LocalTime.ofNanoOfDay(TimeUnit.MILLISECONDS.toNanos((int) val)));
@@ -239,11 +247,16 @@ private static ZonedDateTime getZonedDateTime(long ts, TimeUnit unit) {
     long mod = unit.convert(1, TimeUnit.SECONDS);
     int fraction = (int) (ts % mod);
     long tsInSeconds = unit.toSeconds(ts);
+   return getZonedDateTime(tsInSeconds, fraction, unit);
+  }
+
+  private static ZonedDateTime getZonedDateTime(long epochSecond, long fraction, TimeUnit fractionUnit) {
     // create an Instant with time in seconds and fraction which will be stored as nano seconds.
-    Instant instant = Instant.ofEpochSecond(tsInSeconds, unit.toNanos(fraction));
+    Instant instant = Instant.ofEpochSecond(epochSecond, fractionUnit.toNanos(fraction));
     return ZonedDateTime.ofInstant(instant, ZoneOffset.UTC);
   }
 
+
   private static Object convert(org.apache.kafka.connect.data.Schema schema, Object val) {
     if (val == null) {
       return null;
@@ -302,7 +315,11 @@ public static Schema convert(org.apache.kafka.connect.data.Schema schema) {
         }
         break;
       case STRING:
-        converted = Schema.of(Schema.Type.STRING);
+        if (ZonedTimestamp.SCHEMA_NAME.equals(schema.name())) {
+          converted = Schema.of(Schema.LogicalType.TIMESTAMP_MICROS);
+        } else {
+          converted = Schema.of(Schema.Type.STRING);
+        }
         break;
       case INT8:
       case INT16:

delta-plugins-common/src/test/java/io/cdap/delta/plugin/common/RecordsTest.java

@@ -17,21 +17,30 @@
 package io.cdap.delta.plugin.common;
 
 import io.cdap.cdap.api.data.format.StructuredRecord;
-import org.apache.kafka.connect.data.Schema;
+import io.cdap.cdap.api.data.schema.Schema;
+import io.debezium.time.ZonedTimestamp;
 import org.apache.kafka.connect.data.SchemaBuilder;
 import org.apache.kafka.connect.data.Struct;
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+
 /**
  * Test case for {@link Records} class.
  */
 public class RecordsTest {
 
   @Test
-  public void testConvert() {
+  public void testConvertTinyInt() {
     String fieldName = "priority";
-    Schema dataSchema = SchemaBuilder.struct().name("TinyIntSchema").field(fieldName, Schema.INT16_SCHEMA).build();
+    org.apache.kafka.connect.data.Schema dataSchema = SchemaBuilder
+                                                        .struct()
+                                                        .name("TinyIntSchema")
+                                                        .field(fieldName,
+                                                               org.apache.kafka.connect.data.Schema.INT16_SCHEMA)
+                                                        .build();
     Struct struct = new Struct(dataSchema);
     Short val = 1;
     struct.put(fieldName, val);
@@ -42,4 +51,22 @@ public void testConvert() {
       io.cdap.cdap.api.data.schema.Schema.Type.INT));
     Assert.assertEquals(val.intValue(), (int) convert.get(fieldName));
   }
+
+  @Test
+  public void testConvertTimeStamp() {
+    String fieldName = "timeCreated";
+    org.apache.kafka.connect.data.Schema dataSchema =
+      SchemaBuilder.struct()
+        .name(fieldName).field(fieldName, ZonedTimestamp.schema()).build();
+    Struct struct = new Struct(dataSchema);
+    String val = "2011-12-03T10:15:30.030431+01:00";
+    struct.put(fieldName, val);
+    StructuredRecord converted = Records.convert(struct);
+    Assert.assertEquals(Schema.of(Schema.LogicalType.TIMESTAMP_MICROS),
+                        converted.getSchema().getField(fieldName).getSchema());
+    Assert.assertEquals(converted.getTimestamp(fieldName),
+                        ZonedDateTime.parse(val).withZoneSameInstant(ZoneId.of("UTC")));
+
+
+  }
 }