Backported 778079e (#60) from master

Author: shyiko

src/main/java/com/github/shyiko/mysql/binlog/BinaryLogClient.java

@@ -760,7 +760,8 @@ private byte[] readPacketSplitInChunks(ByteArrayInputStream inputStream, int pac
 
     private void updateClientBinlogFilenameAndPosition(Event event) {
         EventHeader eventHeader = event.getHeader();
-        if (eventHeader.getEventType() == EventType.ROTATE) {
+        EventType eventType = eventHeader.getEventType();
+        if (eventType == EventType.ROTATE) {
             EventData eventData = event.getData();
             RotateEventData rotateEventData;
             if (eventData instanceof EventDeserializer.EventDataWrapper) {
@@ -771,7 +772,9 @@ private void updateClientBinlogFilenameAndPosition(Event event) {
             binlogFilename = rotateEventData.getBinlogFilename();
             binlogPosition = rotateEventData.getBinlogPosition();
         } else
-        if (eventHeader instanceof EventHeaderV4) {
+        // do not update binlogPosition on TABLE_MAP so that in case of reconnect (using a different instance of
+        // client) table mapping cache could be reconstructed before hitting row mutation event
+        if (eventType != EventType.TABLE_MAP && eventHeader instanceof EventHeaderV4) {
             EventHeaderV4 trackableEventHeader = (EventHeaderV4) eventHeader;
             long nextBinlogPosition = trackableEventHeader.getNextPosition();
             if (nextBinlogPosition > 0) {

src/main/java/com/github/shyiko/mysql/binlog/event/EventType.java

@@ -172,7 +172,7 @@ public enum EventType {
      */
     EXT_WRITE_ROWS,
     /**
-     * Describes deleted rows (within a single table).
+     * Describes updated rows (within a single table).
      * Used in case of RBR (5.1.18+).
      */
     EXT_UPDATE_ROWS,
@@ -188,6 +188,12 @@ public enum EventType {
     ANONYMOUS_GTID,
     PREVIOUS_GTIDS;
 
+    public static boolean isRowMutation(EventType eventType) {
+        return EventType.isWrite(eventType) ||
+               EventType.isUpdate(eventType) ||
+               EventType.isDelete(eventType);
+    }
+
     public static boolean isWrite(EventType eventType) {
         return eventType == PRE_GA_WRITE_ROWS ||
                eventType == WRITE_ROWS ||

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

@@ -52,6 +52,7 @@
 import java.sql.SQLException;
 import java.sql.SQLSyntaxErrorException;
 import java.sql.Statement;
+import java.util.AbstractMap;
 import java.util.BitSet;
 import java.util.Calendar;
 import java.util.List;
@@ -61,6 +62,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -333,6 +335,53 @@ public void execute(Statement statement) throws SQLException {
         return result;
     }
 
+    @Test
+    public void testBinlogPositionPointsToTableMapEventUntilTheEndOfLogicalGroup() throws Exception {
+        final AtomicReference<Map.Entry<String, Long>> markHolder = new AtomicReference<Map.Entry<String, Long>>();
+        BinaryLogClient.EventListener markEventListener = new BinaryLogClient.EventListener() {
+
+            private int counter;
+
+            @Override
+            public void onEvent(Event event) {
+                if (EventType.isRowMutation(event.getHeader().getEventType()) && counter++ == 1) {
+                    // coordinates of second insert
+                    markHolder.set(new AbstractMap.SimpleEntry<String, Long>(client.getBinlogFilename(),
+                        client.getBinlogPosition()));
+                }
+            }
+        };
+        client.registerEventListener(markEventListener);
+        try {
+            master.execute(new Callback<Statement>() {
+                @Override
+                public void execute(Statement statement) throws SQLException {
+                    statement.execute("insert into bikini_bottom values('SpongeBob')");
+                    statement.execute("insert into bikini_bottom values('Patrick')");
+                    statement.execute("insert into bikini_bottom values('Squidward')");
+                }
+            });
+            eventListener.waitFor(WriteRowsEventData.class, 3, DEFAULT_TIMEOUT);
+            final BinaryLogClient anotherClient = new BinaryLogClient(slave.hostname, slave.port,
+                slave.username, slave.password);
+            anotherClient.registerLifecycleListener(new TraceLifecycleListener());
+            CountDownEventListener anotherClientEventListener = new CountDownEventListener();
+            anotherClient.registerEventListener(anotherClientEventListener);
+            Map.Entry<String, Long> mark = markHolder.get();
+            anotherClient.setBinlogFilename(mark.getKey());
+            anotherClient.setBinlogPosition(mark.getValue());
+            anotherClient.connect(DEFAULT_TIMEOUT);
+            try {
+                // expecting Patrick & Squidward
+                anotherClientEventListener.waitFor(WriteRowsEventData.class, 2, DEFAULT_TIMEOUT);
+            } finally {
+                anotherClient.disconnect();
+            }
+        } finally {
+            client.unregisterEventListener(markEventListener);
+        }
+    }
+
     @Test(enabled = false)
     public void testUnsupportedColumnTypeDoesNotCauseClientToFail() throws Exception {
         BinaryLogClient.LifecycleListener lifecycleListenerMock = mock(BinaryLogClient.LifecycleListener.class);