From bba800eb968660ce21974757cfbf0840422259a8 Mon Sep 17 00:00:00 2001 From: winger Date: Mon, 17 Dec 2018 16:14:55 +0800 Subject: [PATCH 1/3] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=BC=80=E5=90=AFgtid?= =?UTF-8?q?=E6=97=B6=EF=BC=8C=E5=90=8C=E6=97=B6=E5=BC=80=E5=90=AFtsdb?= =?UTF-8?q?=E5=90=8Egtid=20=E5=B1=9E=E6=80=A7=E4=B8=A2=E5=A4=B1=E7=9A=84?= =?UTF-8?q?=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/parse/inbound/mysql/MysqlEventParser.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java index 35e98f73..4d7b0a64 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java @@ -813,7 +813,9 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE } if (logPosition.getPostion() != null) { - return logPosition.getPostion(); + EntryPosition position = logPosition.getPostion(); + position.setGtid(endPosition.getGtid()); + return position; } else { return null; } From 3465866278ffd1c9da395577f17402ade061a0bc Mon Sep 17 00:00:00 2001 From: winger Date: Fri, 21 Dec 2018 15:18:07 +0800 Subject: [PATCH 2/3] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=BC=80=E5=90=AFgtid?= =?UTF-8?q?=E6=97=B6=EF=BC=8C=E5=90=8C=E6=97=B6=E5=BC=80=E5=90=AFtsdb?= =?UTF-8?q?=E5=90=8Egtid=20=E5=B1=9E=E6=80=A7=E4=B8=A2=E5=A4=B1=E7=9A=84?= =?UTF-8?q?=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/parse/inbound/ErosaConnection.java | 2 +- .../canal/parse/inbound/mysql/LocalBinLogConnection.java | 2 +- .../otter/canal/parse/inbound/mysql/MysqlConnection.java | 8 +++++++- .../canal/parse/inbound/mysql/MysqlEventParser.java | 9 ++++----- 4 files changed, 13 insertions(+), 8 deletions(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/ErosaConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/ErosaConnection.java index 5355e2e1..758e8173 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/ErosaConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/ErosaConnection.java @@ -20,7 +20,7 @@ public interface ErosaConnection { /** * 用于快速数据查找,和dump的区别在于，seek会只给出部分的数据 */ - public void seek(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException; + public void seek(String binlogfilename, Long binlogPosition, String gtid, SinkFunction func) throws IOException; public void dump(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException; diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java index 5cf5bb9d..411a19d6 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java @@ -75,7 +75,7 @@ public class LocalBinLogConnection implements ErosaConnection { return running; } - public void seek(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException { + public void seek(String binlogfilename, Long binlogPosition, String gtid, SinkFunction func) throws IOException { } public void dump(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException { 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 39ec6c62..777923a0 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 @@ -10,6 +10,7 @@ import java.util.List; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; +import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.math.NumberUtils; import org.slf4j.Logger; @@ -118,7 +119,7 @@ public class MysqlConnection implements ErosaConnection { /** * 加速主备切换时的查找速度，做一些特殊优化，比如只解析事务头或者尾 */ - public void seek(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException { + public void seek(String binlogfilename, Long binlogPosition, String gtid, SinkFunction func) throws IOException { updateSettings(); loadBinlogChecksum(); sendBinlogDump(binlogfilename, binlogPosition); @@ -130,6 +131,11 @@ public class MysqlConnection implements ErosaConnection { decoder.handle(LogEvent.QUERY_EVENT); decoder.handle(LogEvent.XID_EVENT); LogContext context = new LogContext(); + // 若entry position存在gtid，则使用传入的gtid作为gtidSet 的拼接标准,否则同时开启gtid和tsdb时，会导致丢失初始化传入的gtid + if (StringUtils.isNotEmpty(gtid)) { + decoder.handle(LogEvent.GTID_LOG_EVENT); + context.setGtidSet(MysqlGTIDSet.parse(gtid)); + } context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum)); while (fetcher.fetch()) { accumulateReceivedBytes(fetcher.limit()); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java index 4d7b0a64..56cdf64f 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java @@ -527,7 +527,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE // 针对开始的第一条为非Begin记录，需要从该binlog扫描 final java.util.concurrent.atomic.AtomicLong preTransactionStartPosition = new java.util.concurrent.atomic.AtomicLong(0L); mysqlConnection.reconnect(); - mysqlConnection.seek(entryPosition.getJournalName(), 4L, new SinkFunction() { + mysqlConnection.seek(entryPosition.getJournalName(), 4L, entryPosition.getGtid(), new SinkFunction() { private LogPosition lastPosition; @@ -740,7 +740,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE try { mysqlConnection.reconnect(); // 开始遍历文件 - mysqlConnection.seek(searchBinlogFile, 4L, new SinkFunction() { + mysqlConnection.seek(searchBinlogFile, 4L, endPosition.getGtid(), new SinkFunction() { private LogPosition lastPosition; @@ -754,6 +754,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE event.getLogPos() - event.getEventLen(), event.getWhen() * 1000, event.getServerId()); + entryPosition.setGtid(event.getHeader().getGtidSetStr()); logPosition.setPostion(entryPosition); } @@ -813,9 +814,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE } if (logPosition.getPostion() != null) { - EntryPosition position = logPosition.getPostion(); - position.setGtid(endPosition.getGtid()); - return position; + return logPosition.getPostion(); } else { return null; } From 19c35c0b086650b4847fdaee306eadd9d3c9623e Mon Sep 17 00:00:00 2001 From: winger Date: Wed, 26 Dec 2018 15:10:57 +0800 Subject: [PATCH 3/3] =?UTF-8?q?=E4=BF=AE=E5=A4=8D:=E5=BD=93mysql=20?= =?UTF-8?q?=E5=8F=91=E7=94=9F=E8=BF=87purged=20gtid=E6=97=B6=EF=BC=8C?= =?UTF-8?q?=E5=90=8C=E6=97=B6=E5=BC=80=E5=90=AFtsdb=E4=B8=8Egtid=E5=90=8E?= =?UTF-8?q?=EF=BC=8Cgtid=E5=B1=9E=E6=80=A7=E4=B8=A2=E5=A4=B1=E9=97=AE?= =?UTF-8?q?=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/parse/inbound/mysql/MysqlConnection.java | 4 +++- .../otter/canal/parse/inbound/mysql/MysqlEventParser.java | 2 ++ 2 files changed, 5 insertions(+), 1 deletion(-) 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 777923a0..afa9543c 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 @@ -131,7 +131,9 @@ public class MysqlConnection implements ErosaConnection { decoder.handle(LogEvent.QUERY_EVENT); decoder.handle(LogEvent.XID_EVENT); LogContext context = new LogContext(); - // 若entry position存在gtid，则使用传入的gtid作为gtidSet 的拼接标准,否则同时开启gtid和tsdb时，会导致丢失初始化传入的gtid + // 若entry position存在gtid，则使用传入的gtid作为gtidSet 拼接的标准,否则同时开启gtid和tsdb时，会导致丢失gtid + // 而当源端数据库gtid 有purged时会有如下类似报错 + // 'errno = 1236, sqlstate = HY000 errmsg = The slave is connecting using CHANGE MASTER TO MASTER_AUTO_POSITION = 1 ... if (StringUtils.isNotEmpty(gtid)) { decoder.handle(LogEvent.GTID_LOG_EVENT); context.setGtidSet(MysqlGTIDSet.parse(gtid)); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java index 56cdf64f..d2ce8356 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java @@ -790,6 +790,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE entryPosition); } logPosition.setPostion(entryPosition); + entryPosition.setGtid(entry.getHeader().getGtid()); } else if (CanalEntry.EntryType.TRANSACTIONBEGIN.equals(entry.getEntryType())) { // 当前事务开始位点 entryPosition = new EntryPosition(logfilename, logfileoffset, logposTimestamp, serverId); @@ -797,6 +798,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE logger.debug("set {} to be pending start position before finding another proper one...", entryPosition); } + entryPosition.setGtid(entry.getHeader().getGtid()); logPosition.setPostion(entryPosition); }