From 3bb408194ddcf39de74af2c94a43537469dc9752 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Thu, 1 Nov 2018 16:03:40 +0800 Subject: [PATCH] fixed issue #1081 , support mariadb rotate_event support checksum --- .../parse/inbound/mysql/MysqlConnection.java | 68 +++++++++++++++++-- .../mysql/MysqlMultiStageCoprocessor.java | 16 ++++- .../canal/parse/DirectLogFetcherTest.java | 48 +++++++++++++ 3 files changed, 123 insertions(+), 9 deletions(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java index 4348dc29..e3056cb9 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java @@ -37,21 +37,23 @@ import com.taobao.tddl.dbsync.binlog.LogBuffer; import com.taobao.tddl.dbsync.binlog.LogContext; import com.taobao.tddl.dbsync.binlog.LogDecoder; import com.taobao.tddl.dbsync.binlog.LogEvent; +import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent; public class MysqlConnection implements ErosaConnection { - private static final Logger logger = LoggerFactory.getLogger(MysqlConnection.class); + private static final Logger logger = LoggerFactory.getLogger(MysqlConnection.class); private MysqlConnector connector; private long slaveId; - private Charset charset = Charset.forName("UTF-8"); + private Charset charset = Charset.forName("UTF-8"); private BinlogFormat binlogFormat; private BinlogImage binlogImage; // tsdb releated private AuthenticationInfo authInfo; - protected int connTimeout = 5 * 1000; // 5秒 - protected int soTimeout = 60 * 60 * 1000; // 1小时 + protected int connTimeout = 5 * 1000; // 5秒 + protected int soTimeout = 60 * 60 * 1000; // 1小时 + private int binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF; // dump binlog bytes, 暂不包括meta与TSDB private AtomicLong receivedBinlogBytes; @@ -118,7 +120,7 @@ public class MysqlConnection implements ErosaConnection { */ public void seek(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException { updateSettings(); - + loadBinlogChecksum(); sendBinlogDump(binlogfilename, binlogPosition); DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize()); fetcher.start(connector.getChannel()); @@ -128,6 +130,7 @@ public class MysqlConnection implements ErosaConnection { decoder.handle(LogEvent.QUERY_EVENT); decoder.handle(LogEvent.XID_EVENT); LogContext context = new LogContext(); + context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum)); while (fetcher.fetch()) { accumulateReceivedBytes(fetcher.limit()); LogEvent event = null; @@ -145,12 +148,14 @@ public class MysqlConnection implements ErosaConnection { public void dump(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException { updateSettings(); + loadBinlogChecksum(); sendRegisterSlave(); sendBinlogDump(binlogfilename, binlogPosition); DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize()); fetcher.start(connector.getChannel()); LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); LogContext context = new LogContext(); + context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum)); while (fetcher.fetch()) { accumulateReceivedBytes(fetcher.limit()); LogEvent event = null; @@ -173,6 +178,7 @@ public class MysqlConnection implements ErosaConnection { @Override public void dump(GTIDSet gtidSet, SinkFunction func) throws IOException { updateSettings(); + loadBinlogChecksum(); sendBinlogDumpGTID(gtidSet); DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize()); @@ -180,6 +186,7 @@ public class MysqlConnection implements ErosaConnection { fetcher.start(connector.getChannel()); LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); LogContext context = new LogContext(); + context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum)); // fix bug: #890 将gtid传输至context中，供decode使用 context.setGtidSet(gtidSet); while (fetcher.fetch()) { @@ -207,9 +214,11 @@ public class MysqlConnection implements ErosaConnection { @Override public void dump(String binlogfilename, Long binlogPosition, MultiStageCoprocessor coprocessor) throws IOException { updateSettings(); + loadBinlogChecksum(); sendRegisterSlave(); sendBinlogDump(binlogfilename, binlogPosition); ((MysqlMultiStageCoprocessor) coprocessor).setConnection(this); + ((MysqlMultiStageCoprocessor) coprocessor).setBinlogChecksum(binlogChecksum); DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize()); try { fetcher.start(connector.getChannel()); @@ -234,9 +243,10 @@ public class MysqlConnection implements ErosaConnection { @Override public void dump(GTIDSet gtidSet, MultiStageCoprocessor coprocessor) throws IOException { updateSettings(); + loadBinlogChecksum(); sendBinlogDumpGTID(gtidSet); - ((MysqlMultiStageCoprocessor) coprocessor).setConnection(this); + ((MysqlMultiStageCoprocessor) coprocessor).setBinlogChecksum(binlogChecksum); DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize()); try { fetcher.start(connector.getChannel()); @@ -482,6 +492,52 @@ public class MysqlConnection implements ErosaConnection { } } + /** + * 获取主库checksum信息 + * + * 
+     * mariadb区别于mysql会在binlog的第一个事件Rotate_Event里也会采用checksum逻辑,而mysql是在第二个binlog事件之后才感知是否需要处理checksum
+     * 导致maraidb只要是开启checksum就会出现binlog文件名解析乱码
+     * fixed issue : https://github.com/alibaba/canal/issues/1081
+     *
+ */ + private void loadBinlogChecksum() { + if (checkMariaDB()) { + ResultSetPacket rs = null; + try { + rs = query("select @@global.binlog_checksum"); + } catch (IOException e) { + throw new CanalParseException(e); + } + + List columnValues = rs.getFieldValues(); + if (columnValues != null && columnValues.size() >= 1 && columnValues.get(0).toUpperCase().equals("CRC32")) { + binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_CRC32; + } else { + binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF; + } + } + } + + /** + * 获取是否为mariadb + */ + private boolean checkMariaDB() { + ResultSetPacket rs = null; + try { + rs = query("SELECT @@version"); + } catch (IOException e) { + throw new CanalParseException(e); + } + + List columnValues = rs.getFieldValues(); + if (columnValues != null && columnValues.size() >= 1) { + return StringUtils.containsIgnoreCase(columnValues.get(0), "MariaDB"); + } + + return false; + } + private void accumulateReceivedBytes(long x) { if (receivedBinlogBytes != null) { receivedBinlogBytes.addAndGet(x); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java index 3db8ecc4..8b682f5d 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java @@ -33,6 +33,7 @@ import com.taobao.tddl.dbsync.binlog.LogContext; import com.taobao.tddl.dbsync.binlog.LogDecoder; import com.taobao.tddl.dbsync.binlog.LogEvent; import com.taobao.tddl.dbsync.binlog.event.DeleteRowsLogEvent; +import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent; import com.taobao.tddl.dbsync.binlog.event.RowsLogEvent; import com.taobao.tddl.dbsync.binlog.event.UpdateRowsLogEvent; import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent; @@ -69,6 +70,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement private WorkerPool workerPool; private BatchEventProcessor simpleParserStage; private BatchEventProcessor sinkStoreStage; + private LogContext logContext; public MysqlMultiStageCoprocessor(int ringBufferSize, int parserThreadCount, LogEventConvert logEventConvert, EventTransactionBuffer transactionBuffer, String destination){ @@ -95,9 +97,10 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement SequenceBarrier sequenceBarrier = disruptorMsgBuffer.newBarrier(); ExceptionHandler exceptionHandler = new SimpleFatalExceptionHandler(); // stage 2 + this.logContext = new LogContext(); simpleParserStage = new BatchEventProcessor(disruptorMsgBuffer, sequenceBarrier, - new SimpleParserStage()); + new SimpleParserStage(logContext)); simpleParserStage.setExceptionHandler(exceptionHandler); disruptorMsgBuffer.addGatingSequences(simpleParserStage.getSequence()); @@ -128,6 +131,12 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement workerPool.start(parserExecutor); } + public void setBinlogChecksum(int binlogChecksum) { + if (binlogChecksum != LogEvent.BINLOG_CHECKSUM_ALG_OFF) { + logContext.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum)); + } + } + @Override public void stop() { // fix bug #968，对于pool与 @@ -239,9 +248,9 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement private LogDecoder decoder; private LogContext context; - public SimpleParserStage(){ + public SimpleParserStage(LogContext context){ decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); - context = new LogContext(); + this.context = context; if (gtidSet != null) { context.setGtidSet(gtidSet); } @@ -468,4 +477,5 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement public void setGtidSet(GTIDSet gtidSet) { this.gtidSet = gtidSet; } + } diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java index aa48f9a3..2cb6ed8b 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java @@ -2,6 +2,7 @@ package com.alibaba.otter.canal.parse; import java.io.IOException; import java.net.InetSocketAddress; +import java.util.List; import org.apache.commons.lang.StringUtils; import org.junit.Assert; @@ -10,21 +11,27 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.parse.driver.mysql.MysqlConnector; +import com.alibaba.otter.canal.parse.driver.mysql.MysqlQueryExecutor; import com.alibaba.otter.canal.parse.driver.mysql.MysqlUpdateExecutor; import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpCommandPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.client.RegisterSlaveCommandPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ErrorPacket; +import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket; import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager; +import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetcher; import com.taobao.tddl.dbsync.binlog.LogContext; import com.taobao.tddl.dbsync.binlog.LogDecoder; import com.taobao.tddl.dbsync.binlog.LogEvent; +import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent; import com.taobao.tddl.dbsync.binlog.event.RotateLogEvent; public class DirectLogFetcherTest { protected final Logger logger = LoggerFactory.getLogger(this.getClass()); + private boolean isMariaDB; + private int binlogChecksum; @Test public void testSimple() { @@ -33,6 +40,7 @@ public class DirectLogFetcherTest { MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "xxxx", "xxxx"); connector.connect(); updateSettings(connector); + loadBinlogChecksum(connector); sendRegisterSlave(connector, 3); sendBinlogDump(connector, "mysql-bin.000001", 4L, 3); @@ -40,6 +48,8 @@ public class DirectLogFetcherTest { LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); LogContext context = new LogContext(); + context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum)); + while (fetcher.fetch()) { LogEvent event = null; event = decoder.decode(fetcher, context); @@ -188,6 +198,44 @@ public class DirectLogFetcherTest { } } + private void loadBinlogChecksum(MysqlConnector connector) { + checkMariaDB(connector); + if (isMariaDB) { + ResultSetPacket rs = null; + try { + rs = query("select @@global.binlog_checksum", connector); + } catch (IOException e) { + throw new CanalParseException(e); + } + + List columnValues = rs.getFieldValues(); + if (columnValues != null && columnValues.size() >= 1 && columnValues.get(0).toUpperCase().equals("CRC32")) { + binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_CRC32; + } else { + binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF; + } + } + } + + private void checkMariaDB(MysqlConnector connector) { + ResultSetPacket rs = null; + try { + rs = query("SELECT @@version", connector); + } catch (IOException e) { + throw new CanalParseException(e); + } + + List columnValues = rs.getFieldValues(); + if (columnValues != null && columnValues.size() >= 1) { + isMariaDB = StringUtils.containsIgnoreCase(columnValues.get(0), "MariaDB"); + } + } + + public ResultSetPacket query(String cmd, MysqlConnector connector) throws IOException { + MysqlQueryExecutor exector = new MysqlQueryExecutor(connector); + return exector.query(cmd); + } + public void update(String cmd, MysqlConnector connector) throws IOException { MysqlUpdateExecutor exector = new MysqlUpdateExecutor(connector); exector.update(cmd);