diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/LogHeader.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/LogHeader.java index 3f677494..a7de734c 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/LogHeader.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/LogHeader.java @@ -326,7 +326,7 @@ public final class LogHeader { if (gtidSet != null) { gtidMap.put(GTID_SET_STRING, gtidSet.toString()); if (gtidEvent != null && gtidEvent instanceof GtidLogEvent) { - GtidLogEvent event = (GtidLogEvent)gtidEvent; + GtidLogEvent event = (GtidLogEvent) gtidEvent; gtidMap.put(CURRENT_GTID_STRING, event.getGtidStr()); gtidMap.put(CURRENT_GTID_SN, String.valueOf(event.getSequenceNumber())); gtidMap.put(CURRENT_GTID_LAST_COMMIT, String.valueOf(event.getLastCommitted())); diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/MariaGTIDSet.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/MariaGTIDSet.java index c9b22c1c..4612b939 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/MariaGTIDSet.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/MariaGTIDSet.java @@ -1,11 +1,11 @@ package com.alibaba.otter.canal.parse.driver.mysql.packets; -import org.apache.commons.lang.StringUtils; - import java.io.IOException; import java.util.HashMap; import java.util.Map; +import org.apache.commons.lang.StringUtils; + /** * 类 MariaGTIDSet.java 的实现 * @@ -13,8 +13,9 @@ import java.util.Map; * @version 1.0.0 */ public class MariaGTIDSet implements GTIDSet { + //MariaDB 10.0.2+ representation of Gtid - Map gtidMap = new HashMap<>(); + private Map gtidMap = new HashMap<>(); @Override public byte[] encode() throws IOException { diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/GtidUtil.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/GtidUtil.java index 3b4cc3d5..6315058a 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/GtidUtil.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/GtidUtil.java @@ -15,7 +15,8 @@ public class GtidUtil { public static GTIDSet parseGtidSet(String gtid, boolean isMariaDB) { if (isMariaDB) { return MariaGTIDSet.parse(gtid); + } else { + return MysqlGTIDSet.parse(gtid); } - return MysqlGTIDSet.parse(gtid); } } 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 c535da4c..67d054a2 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 @@ -1,16 +1,12 @@ package com.alibaba.otter.canal.parse.inbound; +import static com.alibaba.otter.canal.parse.driver.mysql.utils.GtidUtil.parseGtidSet; + import java.io.IOException; -import java.util.Arrays; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Timer; -import java.util.TimerTask; +import java.util.*; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; -import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.apache.commons.lang.math.RandomUtils; @@ -25,7 +21,7 @@ import com.alibaba.otter.canal.parse.CanalEventParser; import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet; import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.parse.exception.PositionNotFoundException; -import com.taobao.tddl.dbsync.binlog.exception.TableIdNotFoundException; +import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection; import com.alibaba.otter.canal.parse.inbound.mysql.MysqlMultiStageCoprocessor; import com.alibaba.otter.canal.parse.index.CanalLogPositionManager; import com.alibaba.otter.canal.parse.support.AuthenticationInfo; @@ -38,8 +34,7 @@ import com.alibaba.otter.canal.protocol.position.LogIdentity; import com.alibaba.otter.canal.protocol.position.LogPosition; import com.alibaba.otter.canal.sink.CanalEventSink; import com.alibaba.otter.canal.sink.exception.CanalSinkException; - -import static com.alibaba.otter.canal.parse.driver.mysql.utils.GtidUtil.parseGtidSet; +import com.taobao.tddl.dbsync.binlog.exception.TableIdNotFoundException; /** * 抽象的EventParser, 最大化共用mysql/oracle版本的实现 @@ -183,7 +178,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle } if (erosaConnection instanceof MysqlConnection) { - isMariaDB = ((MysqlConnection)erosaConnection).isMariaDB(); + isMariaDB = ((MysqlConnection) erosaConnection).isMariaDB(); } // 4. 获取最后的位置信息 long start = System.currentTimeMillis(); @@ -249,7 +244,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle multiStageCoprocessor = buildMultiStageCoprocessor(); if (isGTIDMode() && StringUtils.isNotEmpty(startPosition.getGtid())) { // 判断所属instance是否启用GTID模式，是的话调用ErosaConnection中GTID对应方法dump数据 - GTIDSet gtidSet = parseGtidSet(startPosition.getGtid(),isMariaDB); + GTIDSet gtidSet = parseGtidSet(startPosition.getGtid(), isMariaDB); ((MysqlMultiStageCoprocessor) multiStageCoprocessor).setGtidSet(gtidSet); multiStageCoprocessor.start(); erosaConnection.dump(gtidSet, multiStageCoprocessor); 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 f7de85ec..96d717d5 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 @@ -350,9 +350,9 @@ public class MysqlConnection implements ErosaConnection { private void sendBinlogDumpGTID(GTIDSet gtidSet) throws IOException { if (isMariaDB()) { sendMariaBinlogDumpGTID(gtidSet); - return; + } else { + sendMySQLBinlogDumpGTID(gtidSet); } - sendMySQLBinlogDumpGTID(gtidSet); } private void sendMySQLBinlogDumpGTID(GTIDSet gtidSet) throws IOException { 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 ac31ac29..f510f9af 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 @@ -343,7 +343,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE protected EntryPosition findStartPosition(ErosaConnection connection) throws IOException { if (isGTIDMode()) { - // GTID模式下，CanalLogPositionManager里取最后的gtid，没有则取instanc配置中的 + // GTID模式下，CanalLogPositionManager里取最后的gtid，没有则取instance配置中的 LogPosition logPosition = getLogPositionManager().getLatestIndexBy(destination); if (logPosition != null) { // 如果以前是非GTID模式，后来调整为了GTID模式，那么为了保持兼容，需要判断gtid是否为空 @@ -381,7 +381,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE protected EntryPosition findEndPositionWithMasterIdAndTimestamp(MysqlConnection connection) { MysqlConnection mysqlConnection = (MysqlConnection) connection; final EntryPosition endPosition = findEndPosition(mysqlConnection); - if (tableMetaTSDB != null) { + if (tableMetaTSDB != null || isGTIDMode()) { long startTimestamp = System.currentTimeMillis(); return findAsPerTimestampInSpecificLogFile(mysqlConnection, startTimestamp, @@ -426,7 +426,8 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE } if (entryPosition == null) { - entryPosition = findEndPositionWithMasterIdAndTimestamp(mysqlConnection); // 默认从当前最后一个位置进行消费 + entryPosition = + findEndPositionWithMasterIdAndTimestamp(mysqlConnection); // 默认从当前最后一个位置进行消费 } // 判断一下是否需要按时间订阅