Fix mysql8 signedness decoding

The table metadata only contains signedness bits for integer columns.
So in the case where there's 12 columns but only 1 integer column, mysql
will use just one byte.  Our calculation was saying there should be 2
bytes, and thus our buffer overruneth.

Author: osheroff

pom.xml

@@ -65,7 +65,7 @@
         <dependency>
             <groupId>mysql</groupId>
             <artifactId>mysql-connector-java</artifactId>
-            <version>5.1.21</version>
+            <version>8.0.15</version>
             <scope>test</scope>
         </dependency>
     </dependencies>

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

@@ -46,13 +46,31 @@ public TableMapEventData deserialize(ByteArrayInputStream inputStream) throws IO
         if (metadataLength > 0) {
             metadata = metadataDeserializer.deserialize(
                 new ByteArrayInputStream(inputStream.read(metadataLength)),
-                numberOfColumns
+                numericColumnCount(eventData.getColumnTypes())
             );
         }
         eventData.setEventMetadata(metadata);
         return eventData;
     }
 
+    private int numericColumnCount(byte[] types) {
+        int count = 0;
+        for ( int i = 0 ; i < types.length; i++) {
+            switch ( ColumnType.byCode(types[i] & 0xff)) {
+                case TINY:
+                case SHORT:
+                case INT24:
+                case LONG:
+                case LONGLONG:
+                case NEWDECIMAL:
+                case FLOAT:
+                case DOUBLE:
+                    count++;
+            }
+        }
+        return count;
+    }
+
     private int[] readMetadata(ByteArrayInputStream inputStream, byte[] columnTypes) throws IOException {
         int[] metadata = new int[columnTypes.length];
         for (int i = 0; i < columnTypes.length; i++) {

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

@@ -32,7 +32,7 @@
  */
 public class TableMapEventMetadataDeserializer {
 
-    public TableMapEventMetadata deserialize(ByteArrayInputStream inputStream, int numberOfColumns) throws IOException {
+    public TableMapEventMetadata deserialize(ByteArrayInputStream inputStream, int numberOfIntegerColumns) throws IOException {
         int remainingBytes = inputStream.available();
         if (remainingBytes <= 0) {
             return null;
@@ -49,7 +49,7 @@ public TableMapEventMetadata deserialize(ByteArrayInputStream inputStream, int n
 
             switch (fieldType) {
                 case SIGNEDNESS:
-                    result.setSignedness(readSignedness(inputStream, numberOfColumns));
+                    result.setSignedness(readSignedness(inputStream, numberOfIntegerColumns));
                     break;
                 case DEFAULT_CHARSET:
                     result.setDefaultCharset(readDefaultCharset(inputStream));

src/test/java/com/github/shyiko/mysql/binlog/BinaryLogClientIntegrationTest.java

@@ -990,6 +990,22 @@ public void execute(Statement statement) throws SQLException {
         }
     }
 
+    @Test
+    public void testMysql8TableMetadata() throws Exception {
+        master.execute("create table test_metameta ( " +
+                "a date, b date, c date, d date, e date, f date, g date, " +
+                "h date, i date, j int)");
+        master.execute("insert into test_metameta set j = 5");
+
+        final BinaryLogClient binaryLogClient = new BinaryLogClient(
+            master.hostname, master.port, master.username, master.password
+        );
+        binaryLogClient.registerEventListener(eventListener);
+        binaryLogClient.connect(DEFAULT_TIMEOUT);
+
+        eventListener.waitFor(WriteRowsEventData.class, 1, DEFAULT_TIMEOUT);
+    }
+
     @AfterMethod
     public void afterEachTest() throws Exception {
         final CountDownLatch latch = new CountDownLatch(1);