Merge pull request #5 from fivetran/deserializer-fix-precesion

fix(AbstractRowsEventDataDeserializer): correct fsp accuracy for datetime/timestamp v2

Author: glarwood

src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/AbstractRowsEventDataDeserializer.java

@@ -22,9 +22,9 @@
 import java.io.IOException;
 import java.io.Serializable;
 import java.math.BigDecimal;
+import java.sql.Time;
 import java.util.BitSet;
 import java.util.Calendar;
-import java.util.Date;
 import java.util.Map;
 import java.util.TimeZone;
 
@@ -266,7 +266,7 @@ protected Serializable deserializeTime(ByteArrayInputStream inputStream) throws
         if (deserializeDateAndTimeAsLong) {
             return castTimestamp(timestamp, 0);
         }
-        return timestamp != null ? new java.sql.Time(timestamp) : null;
+        return timestamp != null ? new java.sql.Timestamp(timestamp) : null;
     }
 
     protected Serializable deserializeTimeV2(int meta, ByteArrayInputStream inputStream) throws IOException {
@@ -294,7 +294,7 @@ protected Serializable deserializeTimeV2(int meta, ByteArrayInputStream inputStr
         if (deserializeDateAndTimeAsLong) {
             return castTimestamp(timestamp, fsp);
         }
-        return timestamp != null ? new java.sql.Time(timestamp) : null;
+        return timestamp != null ? convertLongTimestamptWithFSP(timestamp, fsp) : null;
     }
 
     protected Serializable deserializeTimestamp(ByteArrayInputStream inputStream) throws IOException {
@@ -312,7 +312,7 @@ protected Serializable deserializeTimestampV2(int meta, ByteArrayInputStream inp
         if (deserializeDateAndTimeAsLong) {
             return castTimestamp(timestamp, fsp);
         }
-        return new java.sql.Timestamp(timestamp);
+        return convertLongTimestamptWithFSP(timestamp, fsp);
     }
 
     protected Serializable deserializeDatetime(ByteArrayInputStream inputStream) throws IOException {
@@ -321,7 +321,7 @@ protected Serializable deserializeDatetime(ByteArrayInputStream inputStream) thr
         if (deserializeDateAndTimeAsLong) {
             return castTimestamp(timestamp, 0);
         }
-        return timestamp != null ? new java.util.Date(timestamp) : null;
+        return timestamp != null ? new java.sql.Timestamp(timestamp) : null;
     }
 
     protected Serializable deserializeDatetimeV2(int meta, ByteArrayInputStream inputStream) throws IOException {
@@ -354,7 +354,14 @@ protected Serializable deserializeDatetimeV2(int meta, ByteArrayInputStream inpu
         if (deserializeDateAndTimeAsLong) {
             return castTimestamp(timestamp, fsp);
         }
-        return timestamp != null ? new java.util.Date(timestamp) : null;
+
+        return timestamp != null ? convertLongTimestamptWithFSP(timestamp, fsp) : null;
+    }
+
+    private java.sql.Timestamp convertLongTimestamptWithFSP(Long timestamp, int fsp) {
+        java.sql.Timestamp ts = new java.sql.Timestamp(timestamp);
+        ts.setNanos(fsp * 1000);
+        return ts;
     }
 
     protected Serializable deserializeYear(ByteArrayInputStream inputStream) throws IOException {