Added CompatibilityMode.INVALID_DATE_AND_TIME_AS_ZERO (EventDeserializer) (shyiko#210)

Author: shyiko

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

@@ -72,6 +72,7 @@ public abstract class AbstractRowsEventDataDeserializer<T extends EventData> imp
     private final Map<Long, TableMapEventData> tableMapEventByTableId;
 
     private boolean deserializeDateAndTimeAsLong;
+    private boolean deserializeInvalidDateAndTimeAsZero;
     private boolean microsecondsPrecision;
     private boolean deserializeCharAndBinaryAsByteArray;
 
@@ -83,6 +84,10 @@ void setDeserializeDateAndTimeAsLong(boolean value) {
         this.deserializeDateAndTimeAsLong = value;
     }
 
+    void setDeserializeInvalidDateAndTimeAsZero(boolean value) {
+        this.deserializeInvalidDateAndTimeAsZero = value;
+    }
+
     void setMicrosecondsPrecision(boolean value) {
         this.microsecondsPrecision = value;
     }
@@ -409,7 +414,7 @@ protected byte[] deserializeJson(int meta, ByteArrayInputStream inputStream) thr
     protected Long asUnixTime(int year, int month, int day, int hour, int minute, int second, int millis) {
         // https://dev.mysql.com/doc/refman/5.0/en/datetime.html
         if (year == 0 || month == 0 || day == 0) {
-            return null;
+            return deserializeInvalidDateAndTimeAsZero ? 0L : null;
         }
         return UnixTime.from(year, month, day, hour, minute, second, millis);
     }

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

@@ -163,6 +163,9 @@ private void ensureCompatibility(EventDataDeserializer eventDataDeserializer) {
             deserializer.setMicrosecondsPrecision(
                 compatibilitySet.contains(CompatibilityMode.DATE_AND_TIME_AS_LONG_MICRO)
             );
+            deserializer.setDeserializeInvalidDateAndTimeAsZero(
+                compatibilitySet.contains(CompatibilityMode.INVALID_DATE_AND_TIME_AS_ZERO)
+            );
             deserializer.setDeserializeCharAndBinaryAsByteArray(
                 compatibilitySet.contains(CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY)
             );
@@ -226,6 +229,7 @@ public EventDataDeserializer getEventDataDeserializer(EventType eventType) {
     /**
      * @see CompatibilityMode#DATE_AND_TIME_AS_LONG
      * @see CompatibilityMode#DATE_AND_TIME_AS_LONG_MICRO
+     * @see CompatibilityMode#INVALID_DATE_AND_TIME_AS_ZERO
      * @see CompatibilityMode#CHAR_AND_BINARY_AS_BYTE_ARRAY
      */
     public enum CompatibilityMode {
@@ -241,6 +245,13 @@ public enum CompatibilityMode {
          * Same as {@link CompatibilityMode#DATE_AND_TIME_AS_LONG} but values are returned in microseconds.
          */
         DATE_AND_TIME_AS_LONG_MICRO,
+        /**
+         * Return 0 instead of null if year/month/day is 0.
+         * Affects DATETIME/DATETIME_V2/DATE/TIME/TIME_V2.
+         *
+         * <p>This option is going to be enabled by default starting from mysql-binlog-connector-java@1.0.0.
+         */
+        INVALID_DATE_AND_TIME_AS_ZERO,
         /**
          * Return CHAR/VARCHAR/BINARY/VARBINARY values as byte[]|s (instead of String|s).
          *