Fixed rotate event checksum

src/main/java/info/avalon566/shardingscaling/sync/mysql/binlog/MySQLConnector.java

@@ -189,6 +189,7 @@ public final class MySQLConnector {
      */
     public synchronized void subscribe(final String binlogFileName, final long binlogPosition) {
         initDumpConnectSession();
+        int checksumLength = queryChecksumLength();
         registerSlave();
         responseCallback = null;
         BinlogDumpCommandPacket binlogDumpCmd = new BinlogDumpCommandPacket();
@@ -197,11 +198,24 @@ public final class MySQLConnector {
         binlogDumpCmd.setSlaveServerId(serverId);
         channel.pipeline().remove(MySQLCommandPacketDecoder.class);
         channel.pipeline().remove(MySQLCommandResponHandler.class);
-        channel.pipeline().addLast(new MySQLBinlogEventPacketDecoder());
+        channel.pipeline().addLast(new MySQLBinlogEventPacketDecoder(checksumLength));
         channel.pipeline().addLast(new MySQLBinlogEventHandler());
         channel.writeAndFlush(binlogDumpCmd);
     }
 
+    private int queryChecksumLength() {
+        InternalResultSet resultSet = executeQuery("select @@global.binlog_checksum");
+        String checksumType = resultSet.getFieldValues().get(0).getColumns().get(0);
+        switch (checksumType) {
+            case "None":
+                return 0;
+            case "CRC32":
+                return 4;
+            default:
+                throw new UnsupportedOperationException();
+        }
+    }
+
     /**
      * Poll binlog event.
      *

src/main/java/info/avalon566/shardingscaling/sync/mysql/binlog/codec/MySQLBinlogEventPacketDecoder.java

@@ -43,7 +43,12 @@ import java.util.List;
 @Slf4j
 public final class MySQLBinlogEventPacketDecoder extends ByteToMessageDecoder {
 
-    private final BinlogContext binlogContext = new BinlogContext();
+    private final BinlogContext binlogContext;
+
+    public MySQLBinlogEventPacketDecoder(int checksumLength) {
+        binlogContext = new BinlogContext();
+        binlogContext.setChecksumLength(checksumLength);
+    }
 
     @Override
     protected void decode(final ChannelHandlerContext ctx, final ByteBuf in, final List<Object> out) {
@@ -99,7 +104,6 @@ public final class MySQLBinlogEventPacketDecoder extends ByteToMessageDecoder {
 
     private void removeChecksum(final short eventType, final ByteBuf in) {
         if (0 < binlogContext.getChecksumLength()
-                && EventTypes.ROTATE_EVENT != eventType
                 && EventTypes.FORMAT_DESCRIPTION_EVENT != eventType) {
             in.writerIndex(in.writerIndex() - binlogContext.getChecksumLength());
         }