From 7de40091a34aaef868fa1244dc29a46602f625e9 Mon Sep 17 00:00:00 2001 From: winger Date: Mon, 3 Sep 2018 15:28:15 +0800 Subject: [PATCH] fix bug #890 --- .../taobao/tddl/dbsync/binlog/LogContext.java | 15 ++++--------- .../taobao/tddl/dbsync/binlog/LogDecoder.java | 21 ++++++++++++------- .../dbsync/binlog/event/GtidLogEvent.java | 7 +++++++ .../parse/inbound/AbstractEventParser.java | 10 ++++++--- .../mysql/AbstractMysqlEventParser.java | 7 ------- .../parse/inbound/mysql/MysqlConnection.java | 2 ++ .../mysql/MysqlMultiStageCoprocessor.java | 8 +++++++ .../inbound/mysql/dbsync/LogEventConvert.java | 14 +------------ 8 files changed, 42 insertions(+), 42 deletions(-) diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogContext.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogContext.java index 80b2c254..11005197 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogContext.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogContext.java @@ -66,18 +66,11 @@ public final class LogContext { mapOfTable.clear(); } - public final void putGtid(GtidLogEvent logEvent) { - if (logEvent != null) { - String gtid = logEvent.getSid().toString() + ":" + logEvent.getGno(); - if (gtidSet == null) { - gtid = logEvent.getSid().toString() + ":1-" + logEvent.getGno(); - gtidSet = MysqlGTIDSet.parse(gtid); - } - gtidSet.update(gtid); - } - } - public GTIDSet getGtidSet() { return gtidSet; } + + public void setGtidSet(GTIDSet gtidSet) { + this.gtidSet = gtidSet; + } } diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java index d094b2fd..c4a99852 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java @@ -4,6 +4,7 @@ import java.io.IOException; import java.util.BitSet; import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet; +import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -47,22 +48,22 @@ import com.taobao.tddl.dbsync.binlog.event.mariadb.StartEncryptionLogEvent; /** * Implements a binary-log decoder. - * + * * 
  * LogDecoder decoder = new LogDecoder();
  * decoder.handle(...);
- * 
+ *
  * LogEvent event;
  * do
  * {
  *     event = decoder.decode(buffer, context);
- * 
+ *
  *     // process log event.
  * }
  * while (event != null);
  * // no more events in buffer.
  *
- * + * * @author Changyuan.lh * @version 1.0 */ @@ -89,7 +90,7 @@ public final class LogDecoder { /** * Decoding an event from binary-log buffer. - * + * * @return UknownLogEvent if event type is unknown or skipped, * null if buffer is not including a full event. */ @@ -141,7 +142,7 @@ public final class LogDecoder { /** * Deserialize an event from buffer. - * + * * @return UknownLogEvent if event type is unknown or skipped. */ public static LogEvent decode(LogBuffer buffer, LogHeader header, LogContext context) throws IOException { @@ -160,6 +161,7 @@ public final class LogDecoder { // remove checksum bytes buffer.limit(header.getEventLen() - LogEvent.BINLOG_CHECKSUM_LEN); } + GTIDSet gtidSet = context.getGtidSet(); switch (header.getType()) { case LogEvent.QUERY_EVENT: { QueryLogEvent event = new QueryLogEvent(header, buffer, descriptionEvent); @@ -373,8 +375,11 @@ public final class LogDecoder { GtidLogEvent event = new GtidLogEvent(header, buffer, descriptionEvent); /* updating position in context */ logPosition.position = header.getLogPos(); - // update latest gtid - context.putGtid(event); + if (gtidSet != null) { + gtidSet.update(event.getGtidStr()); + // update latest gtid + header.putGtidStr(gtidSet); + } return event; } case LogEvent.PREVIOUS_GTIDS_LOG_EVENT: { diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/GtidLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/GtidLogEvent.java index 5c2678b5..b2897d58 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/GtidLogEvent.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/GtidLogEvent.java @@ -82,4 +82,11 @@ public class GtidLogEvent extends LogEvent { public Long getSequenceNumber() { return sequenceNumber; } + + public String getGtidStr() { + StringBuilder sb = new StringBuilder(); + sb.append(sid.toString()).append(":"); + sb.append(gno); + return sb.toString(); + } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java index fcae1317..be2354dd 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java @@ -8,6 +8,8 @@ import java.util.TimerTask; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet; +import com.alibaba.otter.canal.parse.inbound.mysql.MysqlMultiStageCoprocessor; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.apache.commons.lang.math.RandomUtils; @@ -235,12 +237,14 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle if (parallel) { // build stage processor multiStageCoprocessor = buildMultiStageCoprocessor(); - multiStageCoprocessor.start(); - if (isGTIDMode()) { // 判断所属instance是否启用GTID模式，是的话调用ErosaConnection中GTID对应方法dump数据 - erosaConnection.dump(MysqlGTIDSet.parse(startPosition.getGtid()), multiStageCoprocessor); + GTIDSet gtidSet = MysqlGTIDSet.parse(startPosition.getGtid()); + ((MysqlMultiStageCoprocessor) multiStageCoprocessor).setGtidSet(gtidSet); + multiStageCoprocessor.start(); + erosaConnection.dump(gtidSet, multiStageCoprocessor); } else { + multiStageCoprocessor.start(); if (StringUtils.isEmpty(startPosition.getJournalName()) && startPosition.getTimestamp() != null) { erosaConnection.dump(startPosition.getTimestamp(), multiStageCoprocessor); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java index 941f5ebc..4883d971 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java @@ -76,13 +76,6 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { * @return */ protected boolean processTableMeta(EntryPosition position) { - if (isGTIDMode()) { - if (binlogParser instanceof LogEventConvert) { - // 记录gtid - ((LogEventConvert) binlogParser).setGtidSet(MysqlGTIDSet.parse(position.getGtid())); - } - } - if (tableMetaTSDB != null) { if (position.getTimestamp() == null || position.getTimestamp() <= 0) { throw new CanalParseException("use gtid and TableMeta TSDB should be config timestamp > 0"); 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 eb3c0e31..256b1ecc 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 @@ -179,6 +179,8 @@ public class MysqlConnection implements ErosaConnection { fetcher.start(connector.getChannel()); LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); LogContext context = new LogContext(); + // fix bug: #890 将gtid传输至context中，供decode使用 + context.setGtidSet(gtidSet); while (fetcher.fetch()) { accumulateReceivedBytes(fetcher.limit()); LogEvent event = null; 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 722450dd..89c6e842 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 @@ -7,6 +7,7 @@ import java.util.concurrent.locks.LockSupport; import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; import com.alibaba.otter.canal.common.utils.NamedThreadFactory; +import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet; import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.parse.inbound.ErosaConnection; import com.alibaba.otter.canal.parse.inbound.EventTransactionBuffer; @@ -63,6 +64,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement private String destination; private volatile CanalParseException exception; private AtomicLong eventsPublishBlockingTime; + private GTIDSet gtidSet; public MysqlMultiStageCoprocessor(int ringBufferSize, int parserThreadCount, LogEventConvert logEventConvert, EventTransactionBuffer transactionBuffer, String destination){ @@ -227,6 +229,9 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement public SimpleParserStage(){ decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); context = new LogContext(); + if (gtidSet != null) { + context.setGtidSet(gtidSet); + } } public void onEvent(MessageEvent event, long sequence, boolean endOfBatch) throws Exception { @@ -447,4 +452,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement this.eventsPublishBlockingTime = eventsPublishBlockingTime; } + public void setGtidSet(GTIDSet gtidSet) { + this.gtidSet = gtidSet; + } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java index cb579961..17c55e8d 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java @@ -99,13 +99,6 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar private boolean filterRows = false; private boolean useDruidDdlFilter = true; - // latest gtid - private GTIDSet gtidSet; - - public LogEventConvert(GTIDSet gtidSet){ - this.gtidSet = gtidSet; - } - public LogEventConvert(){ } @@ -172,10 +165,9 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar private Entry parseGTIDLogEvent(GtidLogEvent logEvent) { LogHeader logHeader = logEvent.getHeader(); - String value = logEvent.getSid().toString() + ":" + logEvent.getGno(); Pair.Builder builder = Pair.newBuilder(); builder.setKey("gtid"); - builder.setValue(value); + builder.setValue(logEvent.getGtidStr()); if (logEvent.getLastCommitted() != null) { builder.setKey("lastCommitted"); @@ -974,8 +966,4 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar public void setFilterRows(boolean filterRows) { this.filterRows = filterRows; } - - public void setGtidSet(GTIDSet gtidSet) { - this.gtidSet = gtidSet; - } }