聊聊EventDeserializer的CompatibilityMode

    public enum CompatibilityMode {
        /**
         * Return DATETIME/DATETIME_V2/TIMESTAMP/TIMESTAMP_V2/DATE/TIME/TIME_V2 values as long|s
         * (number of milliseconds since the epoch (00:00:00 Coordinated Universal Time (UTC), Thursday, 1 January 1970,
         * not counting leap seconds)) (instead of java.util.Date/java.sql.Timestamp/java.sql.Date/new java.sql.Time).
         *
         * <p>This option is going to be enabled by default starting from mysql-binlog-connector-java@1.0.0.
         */
        DATE_AND_TIME_AS_LONG,
        /**
         * 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.
         */
        INVALID_DATE_AND_TIME_AS_ZERO,
        /**
         * Return -1 instead of null if year/month/day is 0.
         * Affects DATETIME/DATETIME_V2/DATE/TIME/TIME_V2.
         *
         * @deprecated
         */
        INVALID_DATE_AND_TIME_AS_NEGATIVE_ONE,
        /**
         * Return Long.MIN_VALUE instead of null if year/month/day is 0.
         * Affects DATETIME/DATETIME_V2/DATE/TIME/TIME_V2.
         */
        INVALID_DATE_AND_TIME_AS_MIN_VALUE,
        /**
         * Return CHAR/VARCHAR/BINARY/VARBINARY values as byte[]|s (instead of String|s).
         *
         * <p>This option is going to be enabled by default starting from mysql-binlog-connector-java@1.0.0.
         */
        CHAR_AND_BINARY_AS_BYTE_ARRAY
    }

public class EventDeserializer {

    //......

    private void ensureCompatibility(EventDataDeserializer eventDataDeserializer) {
        if (eventDataDeserializer instanceof AbstractRowsEventDataDeserializer) {
            AbstractRowsEventDataDeserializer deserializer =
                (AbstractRowsEventDataDeserializer) eventDataDeserializer;
            boolean deserializeDateAndTimeAsLong =
                compatibilitySet.contains(CompatibilityMode.DATE_AND_TIME_AS_LONG) ||
                compatibilitySet.contains(CompatibilityMode.DATE_AND_TIME_AS_LONG_MICRO);
            deserializer.setDeserializeDateAndTimeAsLong(deserializeDateAndTimeAsLong);
            deserializer.setMicrosecondsPrecision(
                compatibilitySet.contains(CompatibilityMode.DATE_AND_TIME_AS_LONG_MICRO)
            );
            if (compatibilitySet.contains(CompatibilityMode.INVALID_DATE_AND_TIME_AS_ZERO)) {
                deserializer.setInvalidDateAndTimeRepresentation(0L);
            }
            if (compatibilitySet.contains(CompatibilityMode.INVALID_DATE_AND_TIME_AS_NEGATIVE_ONE)) {
                if (!deserializeDateAndTimeAsLong) {
                    throw new IllegalArgumentException("INVALID_DATE_AND_TIME_AS_NEGATIVE_ONE requires " +
                        "DATE_AND_TIME_AS_LONG or DATE_AND_TIME_AS_LONG_MICRO");
                }
                deserializer.setInvalidDateAndTimeRepresentation(-1L);
            }
            if (compatibilitySet.contains(CompatibilityMode.INVALID_DATE_AND_TIME_AS_MIN_VALUE)) {
                if (!deserializeDateAndTimeAsLong) {
                    throw new IllegalArgumentException("INVALID_DATE_AND_TIME_AS_MIN_VALUE requires " +
                        "DATE_AND_TIME_AS_LONG or DATE_AND_TIME_AS_LONG_MICRO");
                }
                deserializer.setInvalidDateAndTimeRepresentation(Long.MIN_VALUE);
            }
            deserializer.setDeserializeCharAndBinaryAsByteArray(
                compatibilitySet.contains(CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY)
            );
        }
    }

    //......

}

public abstract class AbstractRowsEventDataDeserializer<T extends EventData> implements EventDataDeserializer<T> {

    //......

    private Long castTimestamp(Long timestamp, int fsp) {
        if (microsecondsPrecision && timestamp != null && !timestamp.equals(invalidDateAndTimeRepresentation)) {
            return timestamp * 1000 + fsp % 1000;
        }
        return timestamp;
    }

    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 invalidDateAndTimeRepresentation;
        }
        return UnixTime.from(year, month, day, hour, minute, second, millis);
    }

    //......

}