add EventDeserializer.CompatibilityMode.INTEGER_AS_BYTE_ARRAY impl.

Author: zzt93

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

@@ -75,6 +75,7 @@ public abstract class AbstractRowsEventDataDeserializer<T extends EventData> imp
     private Long invalidDateAndTimeRepresentation;
     private boolean microsecondsPrecision;
     private boolean deserializeCharAndBinaryAsByteArray;
+    private boolean deserializeIntegerAsByteArray;
 
     public AbstractRowsEventDataDeserializer(Map<Long, TableMapEventData> tableMapEventByTableId) {
         this.tableMapEventByTableId = tableMapEventByTableId;
@@ -97,6 +98,10 @@ void setDeserializeCharAndBinaryAsByteArray(boolean value) {
         this.deserializeCharAndBinaryAsByteArray = value;
     }
 
+    void setDeserializeIntegerAsByteArray(boolean deserializeIntegerAsByteArray) {
+        this.deserializeIntegerAsByteArray = deserializeIntegerAsByteArray;
+    }
+
     protected Serializable[] deserializeRow(long tableId, BitSet includedColumns, ByteArrayInputStream inputStream)
             throws IOException {
         TableMapEventData tableMapEvent = tableMapEventByTableId.get(tableId);
@@ -203,22 +208,37 @@ protected Serializable deserializeBit(int meta, ByteArrayInputStream inputStream
     }
 
     protected Serializable deserializeTiny(ByteArrayInputStream inputStream) throws IOException {
+        if (deserializeIntegerAsByteArray) {
+            return inputStream.read(1);
+        }
         return (int) ((byte) inputStream.readInteger(1));
     }
 
     protected Serializable deserializeShort(ByteArrayInputStream inputStream) throws IOException {
+        if (deserializeIntegerAsByteArray) {
+            return inputStream.read(2);
+        }
         return (int) ((short) inputStream.readInteger(2));
     }
 
     protected Serializable deserializeInt24(ByteArrayInputStream inputStream) throws IOException {
+        if (deserializeIntegerAsByteArray) {
+            return inputStream.read(3);
+        }
         return (inputStream.readInteger(3) << 8) >> 8;
     }
 
     protected Serializable deserializeLong(ByteArrayInputStream inputStream) throws IOException {
+        if (deserializeIntegerAsByteArray) {
+            return inputStream.read(4);
+        }
         return inputStream.readInteger(4);
     }
 
     protected Serializable deserializeLongLong(ByteArrayInputStream inputStream) throws IOException {
+        if (deserializeIntegerAsByteArray) {
+            return inputStream.read(8);
+        }
         return inputStream.readLong(8);
     }
 

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

@@ -200,6 +200,9 @@ private void ensureCompatibility(EventDataDeserializer eventDataDeserializer) {
             deserializer.setDeserializeCharAndBinaryAsByteArray(
                 compatibilitySet.contains(CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY)
             );
+            deserializer.setDeserializeIntegerAsByteArray(
+                compatibilitySet.contains(CompatibilityMode.INTEGER_AS_BYTE_ARRAY)
+            );
         }
     }
 
@@ -350,7 +353,11 @@ public enum CompatibilityMode {
          *
          * <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
+        CHAR_AND_BINARY_AS_BYTE_ARRAY,
+        /**
+         * Return TINY/SHORT/INT24/LONG/LONGLONG values as byte[]|s (instead of int|s).
+         */
+        INTEGER_AS_BYTE_ARRAY
     }
 
     /**