From e808c6ec70d5be731d0e8ba18f9eed10a29336c3 Mon Sep 17 00:00:00 2001 From: lulu2panpan Date: Tue, 28 May 2019 00:05:13 +0800 Subject: [PATCH] =?UTF-8?q?AbstractCanalInstance=E7=9A=84subscribeChange?= =?UTF-8?q?=E6=96=B9=E6=B3=95=EF=BC=8C=E5=A2=9E=E5=8A=A0=E5=AF=B9AbstractE?= =?UTF-8?q?ventParser=E7=9A=84=E7=B1=BB=E5=9E=8B=E5=88=A4=E6=96=AD?= =?UTF-8?q?=EF=BC=9BMysqlEventParser=E5=AF=B9Gtid=E7=9A=84=E5=A4=84?= =?UTF-8?q?=E7=90=86=E9=9C=80=E8=A6=81=E8=80=83=E8=99=91=E5=85=BC=E5=AE=B9?= =?UTF-8?q?=E6=80=A7=20(#1824)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * AbstractCanalInstance的subscribeChange方法，增加对AbstractEventParser的类型判断，以支持用户的自定义CanalEventParser * MysqlEventParser对Gtid模式的处理需要考虑兼容性，支持系统平滑的从非gtid模式升级为gtid模式 --- .../canal/instance/core/AbstractCanalInstance.java | 8 ++++++-- .../canal/parse/inbound/AbstractEventParser.java | 4 ++-- .../canal/parse/inbound/mysql/MysqlEventParser.java | 13 ++++++++----- 3 files changed, 16 insertions(+), 9 deletions(-) diff --git a/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/AbstractCanalInstance.java b/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/AbstractCanalInstance.java index b1b64d5b..6993a55a 100644 --- a/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/AbstractCanalInstance.java +++ b/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/AbstractCanalInstance.java @@ -53,10 +53,14 @@ public class AbstractCanalInstance extends AbstractCanalLifeCycle implements Can // 处理group的模式 List eventParsers = ((GroupEventParser) eventParser).getEventParsers(); for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动 - ((AbstractEventParser) singleEventParser).setEventFilter(aviaterFilter); + if(singleEventParser instanceof AbstractEventParser) { + ((AbstractEventParser) singleEventParser).setEventFilter(aviaterFilter); + } } } else { - ((AbstractEventParser) eventParser).setEventFilter(aviaterFilter); + if(eventParser instanceof AbstractEventParser) { + ((AbstractEventParser) eventParser).setEventFilter(aviaterFilter); + } } } 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 73b35df6..409c15d2 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 @@ -242,7 +242,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle if (parallel) { // build stage processor multiStageCoprocessor = buildMultiStageCoprocessor(); - if (isGTIDMode()) { + if (isGTIDMode() && StringUtils.isNotEmpty(startPosition.getGtid())) { // 判断所属instance是否启用GTID模式，是的话调用ErosaConnection中GTID对应方法dump数据 GTIDSet gtidSet = MysqlGTIDSet.parse(startPosition.getGtid()); ((MysqlMultiStageCoprocessor) multiStageCoprocessor).setGtidSet(gtidSet); @@ -260,7 +260,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle } } } else { - if (isGTIDMode()) { + if (isGTIDMode() && StringUtils.isNotEmpty(startPosition.getGtid())) { // 判断所属instance是否启用GTID模式，是的话调用ErosaConnection中GTID对应方法dump数据 erosaConnection.dump(MysqlGTIDSet.parse(startPosition.getGtid()), sinkHandler); } else { 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 d2ce8356..68eb5062 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 @@ -355,11 +355,14 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE // GTID模式下，CanalLogPositionManager里取最后的gtid，没有则取instanc配置中的 LogPosition logPosition = getLogPositionManager().getLatestIndexBy(destination); if (logPosition != null) { - return logPosition.getPostion(); - } - - if (masterPosition != null && StringUtils.isNotEmpty(masterPosition.getGtid())) { - return masterPosition; + // 如果以前是非GTID模式，后来调整为了GTID模式，那么为了保持兼容，需要判断gtid是否为空 + if (StringUtils.isNotEmpty(logPosition.getPostion().getGtid())) { + return logPosition.getPostion(); + } + }else { + if (masterPosition != null && StringUtils.isNotEmpty(masterPosition.getGtid())) { + return masterPosition; + } } }