From c0f2d274d71bb1a2670497c9a249aa6ffdc981a6 Mon Sep 17 00:00:00 2001 From: yinxiu Date: Sat, 18 Mar 2017 11:41:56 +0800 Subject: [PATCH] refactoring log position manager module --- .../instance/core/AbstractCanalInstance.java | 6 +- .../manager/CanalInstanceWithManager.java | 34 ++-- .../parse/inbound/AbstractEventParser.java | 8 +- .../inbound/mysql/LocalBinlogEventParser.java | 4 +- .../index/AbstractLogPositionManager.java | 10 ++ .../index/FailbackLogPositionManager.java | 74 -------- .../index/FailoverLogPositionManager.java | 81 +++++++++ .../index/FileMixedLogPositionManager.java | 164 +++++++++--------- ...onManager.java => LogPositionManager.java} | 9 +- .../parse/index/MemoryLogPositionManager.java | 29 ++-- .../parse/index/MetaLogPositionManager.java | 63 +++---- .../parse/index/MixedLogPositionManager.java | 115 ++++++------ .../index/PeriodMixedLogPositionManager.java | 132 +++++++------- .../index/ZooKeeperLogPositionManager.java | 39 ++--- .../inbound/group/GroupEventPaserTest.java | 15 +- .../inbound/mysql/LocalBinlogDumpTest.java | 23 +-- .../mysql/LocalBinlogEventParserTest.java | 8 +- .../parse/inbound/mysql/MysqlDumpTest.java | 15 +- .../inbound/mysql/MysqlEventParserTest.java | 21 +-- .../stub/AbstractCanalLogPositionManager.java | 8 - 20 files changed, 426 insertions(+), 432 deletions(-) create mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/index/AbstractLogPositionManager.java delete mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/index/FailbackLogPositionManager.java create mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/index/FailoverLogPositionManager.java rename parse/src/main/java/com/alibaba/otter/canal/parse/index/{CanalLogPositionManager.java => LogPositionManager.java} (72%) delete mode 100644 parse/src/test/java/com/alibaba/otter/canal/parse/stub/AbstractCanalLogPositionManager.java 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 9143a626..70a03447 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 @@ -2,6 +2,7 @@ package com.alibaba.otter.canal.instance.core; import java.util.List; +import com.alibaba.otter.canal.parse.index.LogPositionManager; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -16,7 +17,6 @@ import com.alibaba.otter.canal.parse.ha.HeartBeatHAController; import com.alibaba.otter.canal.parse.inbound.AbstractEventParser; import com.alibaba.otter.canal.parse.inbound.group.GroupEventParser; import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser; -import com.alibaba.otter.canal.parse.index.CanalLogPositionManager; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.sink.CanalEventSink; @@ -171,7 +171,7 @@ public class AbstractCanalInstance extends AbstractCanalLifeCycle implements Can if (eventParser instanceof AbstractEventParser) { AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser; // 首先启动log position管理器 - CanalLogPositionManager logPositionManager = abstractEventParser.getLogPositionManager(); + LogPositionManager logPositionManager = abstractEventParser.getLogPositionManager(); if (!logPositionManager.isStart()) { logPositionManager.start(); } @@ -196,7 +196,7 @@ public class AbstractCanalInstance extends AbstractCanalLifeCycle implements Can if (eventParser instanceof AbstractEventParser) { AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser; // 首先启动log position管理器 - CanalLogPositionManager logPositionManager = abstractEventParser.getLogPositionManager(); + LogPositionManager logPositionManager = abstractEventParser.getLogPositionManager(); if (logPositionManager.isStart()) { logPositionManager.stop(); } diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java index dd31dce2..a10140b4 100644 --- a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java @@ -6,6 +6,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; +import com.alibaba.otter.canal.parse.index.*; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -37,12 +38,6 @@ import com.alibaba.otter.canal.parse.inbound.AbstractEventParser; import com.alibaba.otter.canal.parse.inbound.group.GroupEventParser; import com.alibaba.otter.canal.parse.inbound.mysql.LocalBinlogEventParser; import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser; -import com.alibaba.otter.canal.parse.index.CanalLogPositionManager; -import com.alibaba.otter.canal.parse.index.FailbackLogPositionManager; -import com.alibaba.otter.canal.parse.index.MemoryLogPositionManager; -import com.alibaba.otter.canal.parse.index.MetaLogPositionManager; -import com.alibaba.otter.canal.parse.index.PeriodMixedLogPositionManager; -import com.alibaba.otter.canal.parse.index.ZooKeeperLogPositionManager; import com.alibaba.otter.canal.parse.support.AuthenticationInfo; import com.alibaba.otter.canal.protocol.position.EntryPosition; import com.alibaba.otter.canal.sink.entry.EntryEventSink; @@ -343,32 +338,25 @@ public class CanalInstanceWithManager extends AbstractCanalInstance { return haController; } - protected CanalLogPositionManager initLogPositionManager() { + protected LogPositionManager initLogPositionManager() { logger.info("init logPositionPersistManager begin..."); IndexMode indexMode = parameters.getIndexMode(); - CanalLogPositionManager logPositionManager = null; + LogPositionManager logPositionManager; if (indexMode.isMemory()) { logPositionManager = new MemoryLogPositionManager(); } else if (indexMode.isZookeeper()) { - logPositionManager = new ZooKeeperLogPositionManager(); - ((ZooKeeperLogPositionManager) logPositionManager).setZkClientx(getZkclientx()); + logPositionManager = new ZooKeeperLogPositionManager(getZkclientx()); } else if (indexMode.isMixed()) { - logPositionManager = new PeriodMixedLogPositionManager(); - - ZooKeeperLogPositionManager zooKeeperLogPositionManager = new ZooKeeperLogPositionManager(); - zooKeeperLogPositionManager.setZkClientx(getZkclientx()); - ((PeriodMixedLogPositionManager) logPositionManager).setZooKeeperLogPositionManager(zooKeeperLogPositionManager); + MemoryLogPositionManager memoryLogPositionManager = new MemoryLogPositionManager(); + ZooKeeperLogPositionManager zooKeeperLogPositionManager = new ZooKeeperLogPositionManager(getZkclientx()); + logPositionManager = new PeriodMixedLogPositionManager(memoryLogPositionManager, zooKeeperLogPositionManager, 1000L); } else if (indexMode.isMeta()) { - logPositionManager = new MetaLogPositionManager(); - ((MetaLogPositionManager) logPositionManager).setMetaManager(metaManager); + logPositionManager = new MetaLogPositionManager(metaManager); } else if (indexMode.isMemoryMetaFailback()) { - MemoryLogPositionManager primaryLogPositionManager = new MemoryLogPositionManager(); - MetaLogPositionManager failbackLogPositionManager = new MetaLogPositionManager(); - failbackLogPositionManager.setMetaManager(metaManager); + MemoryLogPositionManager primary = new MemoryLogPositionManager(); + MetaLogPositionManager secondary = new MetaLogPositionManager(metaManager); - logPositionManager = new FailbackLogPositionManager(); - ((FailbackLogPositionManager) logPositionManager).setPrimary(primaryLogPositionManager); - ((FailbackLogPositionManager) logPositionManager).setFailback(failbackLogPositionManager); + logPositionManager = new FailoverLogPositionManager(primary, secondary); } else { throw new CanalException("unsupport indexMode for " + indexMode); } 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 59fa2839..270f739b 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,7 @@ import java.util.TimerTask; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import com.alibaba.otter.canal.parse.index.LogPositionManager; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.apache.commons.lang.math.RandomUtils; @@ -22,7 +23,6 @@ import com.alibaba.otter.canal.parse.CanalEventParser; import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.parse.exception.TableIdNotFoundException; import com.alibaba.otter.canal.parse.inbound.EventTransactionBuffer.TransactionFlushCallback; -import com.alibaba.otter.canal.parse.index.CanalLogPositionManager; import com.alibaba.otter.canal.parse.support.AuthenticationInfo; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.CanalEntry.Entry; @@ -44,7 +44,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle protected final Logger logger = LoggerFactory.getLogger(this.getClass()); - protected CanalLogPositionManager logPositionManager = null; + protected LogPositionManager logPositionManager = null; protected CanalEventSink> eventSink = null; protected CanalEventFilter eventFilter = null; protected CanalEventFilter eventBlackFilter = null; @@ -495,7 +495,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle return this.alarmHandler; } - public void setLogPositionManager(CanalLogPositionManager logPositionManager) { + public void setLogPositionManager(LogPositionManager logPositionManager) { this.logPositionManager = logPositionManager; } @@ -503,7 +503,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle this.transactionSize = transactionSize; } - public CanalLogPositionManager getLogPositionManager() { + public LogPositionManager getLogPositionManager() { return logPositionManager; } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java index 331b4d45..6334c464 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java @@ -2,6 +2,7 @@ package com.alibaba.otter.canal.parse.inbound.mysql; import java.io.IOException; +import com.alibaba.otter.canal.parse.index.LogPositionManager; import org.apache.commons.lang.StringUtils; import com.alibaba.otter.canal.parse.CanalEventParser; @@ -9,7 +10,6 @@ import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.parse.inbound.ErosaConnection; import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert; import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache; -import com.alibaba.otter.canal.parse.index.CanalLogPositionManager; import com.alibaba.otter.canal.parse.support.AuthenticationInfo; import com.alibaba.otter.canal.protocol.position.EntryPosition; import com.alibaba.otter.canal.protocol.position.LogPosition; @@ -150,7 +150,7 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements // ========================= setter / getter ========================= - public void setLogPositionManager(CanalLogPositionManager logPositionManager) { + public void setLogPositionManager(LogPositionManager logPositionManager) { this.logPositionManager = logPositionManager; } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/AbstractLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/AbstractLogPositionManager.java new file mode 100644 index 00000000..914e4c00 --- /dev/null +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/AbstractLogPositionManager.java @@ -0,0 +1,10 @@ +package com.alibaba.otter.canal.parse.index; + +import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; + +/** + * Created by yinxiu on 17/3/17. + * Email: marklin.hz@gmail.com + */ +public abstract class AbstractLogPositionManager extends AbstractCanalLifeCycle implements LogPositionManager { +} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/FailbackLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/FailbackLogPositionManager.java deleted file mode 100644 index 6fa8efa5..00000000 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/FailbackLogPositionManager.java +++ /dev/null @@ -1,74 +0,0 @@ -package com.alibaba.otter.canal.parse.index; - -import org.springframework.util.Assert; - -import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; -import com.alibaba.otter.canal.parse.exception.CanalParseException; -import com.alibaba.otter.canal.protocol.position.LogPosition; - -/** - * 实现基于failover查找的机制完成meta的操作 - * - * 
- * 应用场景：比如针对内存buffer，出现HA切换，先尝试从内存buffer区中找到lastest position，如果不存在才尝试找一下meta里消费的信息
- *
- * - * @author jianghang 2012-7-20 下午02:33:20 - */ -public class FailbackLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager { - - private CanalLogPositionManager primary; - private CanalLogPositionManager failback; - - public void start() { - super.start(); - Assert.notNull(primary); - Assert.notNull(failback); - - if (!primary.isStart()) { - primary.start(); - } - - if (!failback.isStart()) { - failback.start(); - } - } - - public void stop() { - super.stop(); - - if (primary.isStart()) { - primary.stop(); - } - - if (failback.isStart()) { - failback.stop(); - } - } - - public LogPosition getLatestIndexBy(String destination) { - LogPosition logPosition = primary.getLatestIndexBy(destination); - if (logPosition == null) { - return failback.getLatestIndexBy(destination); - } else { - return logPosition; - } - } - - public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException { - try { - primary.persistLogPosition(destination, logPosition); - } catch (CanalParseException e) { - failback.persistLogPosition(destination, logPosition); - } - } - - public void setPrimary(CanalLogPositionManager primary) { - this.primary = primary; - } - - public void setFailback(CanalLogPositionManager failback) { - this.failback = failback; - } - -} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/FailoverLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/FailoverLogPositionManager.java new file mode 100644 index 00000000..6f36aeae --- /dev/null +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/FailoverLogPositionManager.java @@ -0,0 +1,81 @@ +package com.alibaba.otter.canal.parse.index; + +import com.alibaba.otter.canal.parse.exception.CanalParseException; +import com.alibaba.otter.canal.protocol.position.LogPosition; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Created by yinxiu on 17/3/18. + * Email: marklin.hz@gmail.com + * + * 实现基于failover查找的机制完成meta的操作 + * + * 
+ * 应用场景：比如针对内存buffer，出现HA切换，先尝试从内存buffer区中找到lastest position，如果不存在才尝试找一下meta里消费的信息
+ *
+ */ +public class FailoverLogPositionManager extends AbstractLogPositionManager { + + private final static Logger logger = LoggerFactory.getLogger(FailoverLogPositionManager.class); + + private final LogPositionManager primary; + private final LogPositionManager secondary; + + public FailoverLogPositionManager(LogPositionManager primary, LogPositionManager secondary) { + if (primary == null) { + throw new NullPointerException("nul primary LogPositionManager"); + } + if (secondary == null) { + throw new NullPointerException("nul secondary LogPositionManager"); + } + + this.primary = primary; + this.secondary = secondary; + } + + @Override + public void start() { + super.start(); + + if (!primary.isStart()) { + primary.start(); + } + + if (!secondary.isStart()) { + primary.start(); + } + } + + @Override + public void stop() { + super.stop(); + + if (secondary.isStart()) { + secondary.stop(); + } + + if (primary.isStart()) { + primary.stop(); + } + } + + @Override + public LogPosition getLatestIndexBy(String destination) { + LogPosition logPosition = primary.getLatestIndexBy(destination); + if (logPosition != null) { + return logPosition; + } + return secondary.getLatestIndexBy(destination); + } + + @Override + public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException { + try { + primary.persistLogPosition(destination, logPosition); + } catch (CanalParseException e) { + logger.warn("persistLogPosition use primary log position manager exception. destination: {}, logPosition: {}", destination, logPosition, e); + secondary.persistLogPosition(destination, logPosition); + } + } +} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java index e2c15401..072721af 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java @@ -1,60 +1,82 @@ package com.alibaba.otter.canal.parse.index; +import com.alibaba.otter.canal.common.utils.JsonUtils; +import com.alibaba.otter.canal.meta.exception.CanalMetaManagerException; +import com.alibaba.otter.canal.parse.exception.CanalParseException; +import com.alibaba.otter.canal.protocol.position.LogPosition; +import com.google.common.base.Function; +import com.google.common.collect.MigrateMap; +import org.apache.commons.io.FileUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.io.File; import java.io.IOException; import java.nio.charset.Charset; -import java.util.ArrayList; -import java.util.Collections; -import java.util.HashSet; -import java.util.List; -import java.util.Map; -import java.util.Set; +import java.util.*; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import org.apache.commons.io.FileUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.util.Assert; - -import com.alibaba.otter.canal.common.utils.JsonUtils; -import com.alibaba.otter.canal.meta.exception.CanalMetaManagerException; -import com.alibaba.otter.canal.protocol.position.LogPosition; -import com.google.common.base.Function; -import com.google.common.collect.MigrateMap; - /** + * Created by yinxiu on 17/3/18. + * Email: marklin.hz@gmail.com + * * 基于文件刷新的log position实现 - * + * * 
  * 策略：
  * 1. 先写内存，然后定时刷新数据到File
  * 2. 数据采取overwrite模式(只保留最后一次)
  *
- * - * @author jianghang 2013-4-15 下午09:40:48 - * @version 1.0.4 */ -public class FileMixedLogPositionManager extends MemoryLogPositionManager { +public class FileMixedLogPositionManager extends AbstractLogPositionManager { - private static final Logger logger = LoggerFactory.getLogger(FileMixedLogPositionManager.class); - private static final Charset charset = Charset.forName("UTF-8"); - private File dataDir; - private String dataFileName = "parse.dat"; - private Map dataFileCaches; - private ScheduledExecutorService executor; - @SuppressWarnings("serial") - private final LogPosition nullPosition = new LogPosition() { - }; + private final static Logger logger = LoggerFactory.getLogger(FileMixedLogPositionManager.class); + private final static Charset charset = Charset.forName("UTF-8"); - private long period = 1000; // 单位ms - private Set persistTasks; + private File dataDir; + private Map dataFileCaches; + + private ScheduledExecutorService executorService; + + private final LogPosition nullPosition = new LogPosition(){}; + + private MemoryLogPositionManager memoryLogPositionManager; + + private long period; + private Set persistTasks; + + public FileMixedLogPositionManager(File dataDir, long period, MemoryLogPositionManager memoryLogPositionManager) { + if (dataDir == null) { + throw new NullPointerException("null dataDir"); + } + if (period <= 0) { + throw new IllegalArgumentException("period must be positive, given: " + period); + } + if (memoryLogPositionManager == null) { + throw new NullPointerException("null memoryLogPositionManager"); + } + this.dataDir = dataDir; + this.period = period; + this.memoryLogPositionManager = memoryLogPositionManager; + + this.dataFileCaches = MigrateMap.makeComputingMap(new Function() { + + public File apply(String destination) { + return getDataFile(destination); + } + }); + + this.executorService = Executors.newScheduledThreadPool(1); + this.persistTasks = Collections.synchronizedSet(new HashSet()); + } + + @Override public void start() { super.start(); - Assert.notNull(dataDir); if (!dataDir.exists()) { try { FileUtils.forceMkdir(dataDir); @@ -67,30 +89,12 @@ public class FileMixedLogPositionManager extends MemoryLogPositionManager { throw new CanalMetaManagerException("dir[" + dataDir.getPath() + "] can not read/write"); } - dataFileCaches = MigrateMap.makeComputingMap(new Function() { - - public File apply(String destination) { - return getDataFile(destination); - } - }); - - executor = Executors.newScheduledThreadPool(1); - positions = MigrateMap.makeComputingMap(new Function() { - - public LogPosition apply(String destination) { - LogPosition logPosition = loadDataFromFile(dataFileCaches.get(destination)); - if (logPosition == null) { - return nullPosition; - } else { - return logPosition; - } - } - }); - - persistTasks = Collections.synchronizedSet(new HashSet()); + if (!memoryLogPositionManager.isStart()) { + memoryLogPositionManager.start(); + } // 启动定时工作任务 - executor.scheduleAtFixedRate(new Runnable() { + executorService.scheduleAtFixedRate(new Runnable() { public void run() { List tasks = new ArrayList(persistTasks); @@ -106,28 +110,35 @@ public class FileMixedLogPositionManager extends MemoryLogPositionManager { } } }, period, period, TimeUnit.MILLISECONDS); + } + @Override public void stop() { super.stop(); flushDataToFile(); - executor.shutdownNow(); - positions.clear(); - } - - public void persistLogPosition(String destination, LogPosition logPosition) { - persistTasks.add(destination);// 添加到任务队列中进行触发 - super.persistLogPosition(destination, logPosition); + executorService.shutdown(); + memoryLogPositionManager.stop(); } + @Override public LogPosition getLatestIndexBy(String destination) { - LogPosition logPostion = super.getLatestIndexBy(destination); - if (logPostion == nullPosition) { - return null; - } else { - return logPostion; + LogPosition logPosition = memoryLogPositionManager.getLatestIndexBy(destination); + if (logPosition != null) { + return logPosition; } + logPosition = loadDataFromFile(dataFileCaches.get(destination)); + if (logPosition == null) { + return nullPosition; + } + return logPosition; + } + + @Override + public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException { + persistTasks.add(destination); + memoryLogPositionManager.persistLogPosition(destination, logPosition); } // ============================ helper method ====================== @@ -142,11 +153,12 @@ public class FileMixedLogPositionManager extends MemoryLogPositionManager { } } + String dataFileName = "parse.dat"; return new File(destinationMetaDir, dataFileName); } private void flushDataToFile() { - for (String destination : positions.keySet()) { + for (String destination : memoryLogPositionManager.destinations()) { flushDataToFile(destination); } } @@ -156,7 +168,7 @@ public class FileMixedLogPositionManager extends MemoryLogPositionManager { } private void flushDataToFile(String destination, File dataFile) { - LogPosition position = positions.get(destination); + LogPosition position = memoryLogPositionManager.getLatestIndexBy(destination); if (position != null && position != nullPosition) { String json = JsonUtils.marshalToString(position); try { @@ -179,16 +191,4 @@ public class FileMixedLogPositionManager extends MemoryLogPositionManager { throw new CanalMetaManagerException(e); } } - - public void setDataDir(String dataDir) { - this.dataDir = new File(dataDir); - } - - public void setDataDir(File dataDir) { - this.dataDir = dataDir; - } - - public void setPeriod(long period) { - this.period = period; - } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/CanalLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/LogPositionManager.java similarity index 72% rename from parse/src/main/java/com/alibaba/otter/canal/parse/index/CanalLogPositionManager.java rename to parse/src/main/java/com/alibaba/otter/canal/parse/index/LogPositionManager.java index 25f63d15..3edd0968 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/CanalLogPositionManager.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/LogPositionManager.java @@ -5,14 +5,13 @@ import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.protocol.position.LogPosition; /** - * 接口组合 - * - * @author jianghang 2012-7-7 上午10:02:02 - * @version 1.0.0 + * Created by yinxiu on 17/3/17. + * Email: marklin.hz@gmail.com */ -public interface CanalLogPositionManager extends CanalLifeCycle { +public interface LogPositionManager extends CanalLifeCycle { LogPosition getLatestIndexBy(String destination); void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException; + } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/MemoryLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/MemoryLogPositionManager.java index c95fa4e2..9e20de5a 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/MemoryLogPositionManager.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/MemoryLogPositionManager.java @@ -1,39 +1,44 @@ package com.alibaba.otter.canal.parse.index; -import java.util.Map; - -import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; +import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.protocol.position.LogPosition; import com.google.common.collect.MapMaker; +import java.util.Map; +import java.util.Set; + /** - * 基于内存的实现 - * - * @author jianghang 2012-7-7 上午10:17:23 - * @version 1.0.0 + * Created by yinxiu on 17/3/17. + * Email: marklin.hz@gmail.com */ -public class MemoryLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager { +public class MemoryLogPositionManager extends AbstractLogPositionManager { - protected Map positions; + private Map positions; + @Override public void start() { super.start(); - positions = new MapMaker().makeMap(); } + @Override public void stop() { super.stop(); - positions.clear(); } + @Override public LogPosition getLatestIndexBy(String destination) { return positions.get(destination); } - public void persistLogPosition(String destination, LogPosition logPosition) { + @Override + public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException { positions.put(destination, logPosition); } + public Set destinations() { + return positions.keySet(); + } + } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/MetaLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/MetaLogPositionManager.java index 3af97411..f3dd5479 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/MetaLogPositionManager.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/MetaLogPositionManager.java @@ -1,55 +1,59 @@ package com.alibaba.otter.canal.parse.index; -import java.util.List; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.util.Assert; -import org.springframework.util.CollectionUtils; - -import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; import com.alibaba.otter.canal.meta.CanalMetaManager; +import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.position.LogPosition; import com.alibaba.otter.canal.store.helper.CanalEventUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.util.CollectionUtils; + +import java.util.List; /** - * 基于meta信息的实现 - * - * @author jianghang 2012-7-10 下午05:02:33 - * @version 1.0.0 + * Created by yinxiu on 17/3/18. + * Email: marklin.hz@gmail.com */ -public class MetaLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager { +public class MetaLogPositionManager extends AbstractLogPositionManager { - private static final Logger logger = LoggerFactory.getLogger(MetaLogPositionManager.class); - private CanalMetaManager metaManager; + private final static Logger logger = LoggerFactory.getLogger(MetaLogPositionManager.class); - public void start() { - super.start(); - Assert.notNull(metaManager); - if (!metaManager.isStart()) { - metaManager.start(); + private final CanalMetaManager metaManager; + + public MetaLogPositionManager(CanalMetaManager metaManager) { + if (metaManager == null) { + throw new NullPointerException("null metaManager"); } + + this.metaManager = metaManager; } + @Override public void stop() { super.stop(); + if (metaManager.isStart()) { metaManager.stop(); } } - public void persistLogPosition(String destination, LogPosition logPosition) { - // do nothing - logger.info("persist LogPosition:{}", destination, logPosition); + @Override + public void start() { + super.start(); + + if (!metaManager.isStart()) { + metaManager.start(); + } } + @Override public LogPosition getLatestIndexBy(String destination) { - List clientIdentitys = metaManager.listAllSubscribeInfo(destination); + List clientIdentities = metaManager.listAllSubscribeInfo(destination); LogPosition result = null; - if (!CollectionUtils.isEmpty(clientIdentitys)) { + if (!CollectionUtils.isEmpty(clientIdentities)) { // 尝试找到一个最小的logPosition - for (ClientIdentity clientIdentity : clientIdentitys) { + for (ClientIdentity clientIdentity : clientIdentities) { LogPosition position = (LogPosition) metaManager.getCursor(clientIdentity); if (position == null) { continue; @@ -66,8 +70,9 @@ public class MetaLogPositionManager extends AbstractCanalLifeCycle implements Ca return result; } - public void setMetaManager(CanalMetaManager metaManager) { - this.metaManager = metaManager; + @Override + public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException { + // do nothing + logger.info("persist LogPosition:{}", destination, logPosition); } - } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManager.java index 0016bb06..1e36c6e7 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManager.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManager.java @@ -1,90 +1,83 @@ package com.alibaba.otter.canal.parse.index; +import com.alibaba.otter.canal.common.zookeeper.ZkClientx; +import com.alibaba.otter.canal.parse.exception.CanalParseException; +import com.alibaba.otter.canal.protocol.position.LogPosition; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.util.Assert; - -import com.alibaba.otter.canal.protocol.position.LogPosition; -import com.google.common.base.Function; -import com.google.common.collect.MigrateMap; - /** - * 混合memory + zookeeper的存储模式 - * - * @author jianghang 2012-7-7 上午10:33:19 - * @version 1.0.0 + * Created by yinxiu on 17/3/17. + * Email: marklin.hz@gmail.com + * + * Memory first. + * Asynchronous commit position info to ZK. */ -public class MixedLogPositionManager extends MemoryLogPositionManager implements CanalLogPositionManager { +public class MixedLogPositionManager extends AbstractLogPositionManager { - private static final Logger logger = LoggerFactory.getLogger(MixedLogPositionManager.class); - private ZooKeeperLogPositionManager zooKeeperLogPositionManager; - private ExecutorService executor; - @SuppressWarnings("serial") - private final LogPosition nullPosition = new LogPosition() { - }; + private final Logger logger = LoggerFactory.getLogger(MixedLogPositionManager.class); + private final MemoryLogPositionManager memoryLogPositionManager; + private final ZooKeeperLogPositionManager zooKeeperLogPositionManager; + + private final ExecutorService executor; + + public MixedLogPositionManager(ZkClientx zkClient) { + if (zkClient == null) { + throw new NullPointerException("null zkClient"); + } + + this.memoryLogPositionManager = new MemoryLogPositionManager(); + this.zooKeeperLogPositionManager = new ZooKeeperLogPositionManager(zkClient); + + this.executor = Executors.newFixedThreadPool(1); + } + + @Override public void start() { super.start(); - Assert.notNull(zooKeeperLogPositionManager); - if (!zooKeeperLogPositionManager.isStart()) { - zooKeeperLogPositionManager.start(); - } - executor = Executors.newFixedThreadPool(1); - positions = MigrateMap.makeComputingMap(new Function() { - - public LogPosition apply(String destination) { - LogPosition logPosition = zooKeeperLogPositionManager.getLatestIndexBy(destination); - if (logPosition == null) { - return nullPosition; - } else { - return logPosition; - } - } - }); + memoryLogPositionManager.start(); + zooKeeperLogPositionManager.start(); } + @Override public void stop() { super.stop(); - if (zooKeeperLogPositionManager.isStart()) { - zooKeeperLogPositionManager.stop(); - } - executor.shutdownNow(); - positions.clear(); + executor.shutdown(); + zooKeeperLogPositionManager.stop(); + memoryLogPositionManager.stop(); } - public void persistLogPosition(final String destination, final LogPosition logPosition) { - super.persistLogPosition(destination, logPosition); - executor.submit(new Runnable() { + @Override + public LogPosition getLatestIndexBy(String destination) { + LogPosition logPosition = memoryLogPositionManager.getLatestIndexBy(destination); + if (logPosition != null) { + return logPosition; + } + logPosition = zooKeeperLogPositionManager.getLatestIndexBy(destination); + // 这里保持和重构前的逻辑一致,重新添加到Memory中 + if (logPosition != null) { + memoryLogPositionManager.persistLogPosition(destination, logPosition); + } + return logPosition; + } + @Override + public void persistLogPosition(final String destination, final LogPosition logPosition) throws CanalParseException { + memoryLogPositionManager.persistLogPosition(destination, logPosition); + executor.submit(new Runnable() { public void run() { try { zooKeeperLogPositionManager.persistLogPosition(destination, logPosition); } catch (Exception e) { - logger.error("ERROR # persist to zookeepr has an error", e); + logger.error("ERROR # persist to zookeeper has an error", e); } } }); - } - - public LogPosition getLatestIndexBy(String destination) { - LogPosition logPosition = super.getLatestIndexBy(destination); - if (logPosition == nullPosition) { - return null; - } else { - return logPosition; - } - } - - // ======================== setter / getter ====================== - - public void setZooKeeperLogPositionManager(ZooKeeperLogPositionManager zooKeeperLogPositionManager) { - this.zooKeeperLogPositionManager = zooKeeperLogPositionManager; - } - } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java index e947720a..63a1dae3 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java @@ -1,64 +1,81 @@ package com.alibaba.otter.canal.parse.index; -import java.util.ArrayList; -import java.util.Collections; -import java.util.HashSet; -import java.util.List; -import java.util.Set; +import com.alibaba.otter.canal.parse.exception.CanalParseException; +import com.alibaba.otter.canal.protocol.position.LogPosition; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.*; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.util.Assert; - -import com.alibaba.otter.canal.protocol.position.LogPosition; -import com.google.common.base.Function; -import com.google.common.collect.MigrateMap; - /** - * 基于定时刷新的策略的mixed实现 - * - * @author jianghang 2012-9-12 上午11:18:14 - * @version 1.0.0 + * Created by yinxiu on 17/3/18. + * Email: marklin.hz@gmail.com */ -public class PeriodMixedLogPositionManager extends MemoryLogPositionManager implements CanalLogPositionManager { +public class PeriodMixedLogPositionManager extends AbstractLogPositionManager { - private static final Logger logger = LoggerFactory.getLogger(PeriodMixedLogPositionManager.class); + private static final Logger logger = LoggerFactory.getLogger(PeriodMixedLogPositionManager.class); + + private MemoryLogPositionManager memoryLogPositionManager; private ZooKeeperLogPositionManager zooKeeperLogPositionManager; - private ScheduledExecutorService executor; - @SuppressWarnings("serial") - private final LogPosition nullPosition = new LogPosition() { - }; + private ScheduledExecutorService executorService; - private long period = 1000; // 单位ms - private Set persistTasks; + private long period; + private Set persistTasks; + private final LogPosition nullPosition = new LogPosition(){}; + + public PeriodMixedLogPositionManager(MemoryLogPositionManager memoryLogPositionManager, ZooKeeperLogPositionManager zooKeeperLogPositionManager, long period) { + if (memoryLogPositionManager == null) { + throw new NullPointerException("null memoryLogPositionManager"); + } + + if (zooKeeperLogPositionManager == null) { + throw new NullPointerException("null zooKeeperLogPositionManager"); + } + + if (period <= 0) { + throw new IllegalArgumentException("period must be positive, given: " + period); + } + + this.memoryLogPositionManager = memoryLogPositionManager; + this.zooKeeperLogPositionManager = zooKeeperLogPositionManager; + this.period = period; + this.persistTasks = Collections.synchronizedSet(new HashSet()); + this.executorService = Executors.newScheduledThreadPool(1); + } + + @Override + public void stop() { + super.stop(); + + if (zooKeeperLogPositionManager.isStart()) { + zooKeeperLogPositionManager.stop(); + } + + if (memoryLogPositionManager.isStart()) { + memoryLogPositionManager.stop(); + } + + executorService.shutdown(); + } + + @Override public void start() { super.start(); - Assert.notNull(zooKeeperLogPositionManager); + if (!memoryLogPositionManager.isStart()) { + memoryLogPositionManager.start(); + } + if (!zooKeeperLogPositionManager.isStart()) { zooKeeperLogPositionManager.start(); } - executor = Executors.newScheduledThreadPool(1); - positions = MigrateMap.makeComputingMap(new Function() { - - public LogPosition apply(String destination) { - LogPosition logPosition = zooKeeperLogPositionManager.getLatestIndexBy(destination); - if (logPosition == null) { - return nullPosition; - } else { - return logPosition; - } - } - }); - - persistTasks = Collections.synchronizedSet(new HashSet()); // 启动定时工作任务 - executor.scheduleAtFixedRate(new Runnable() { + executorService.scheduleAtFixedRate(new Runnable() { public void run() { List tasks = new ArrayList(persistTasks); @@ -76,36 +93,19 @@ public class PeriodMixedLogPositionManager extends MemoryLogPositionManager impl }, period, period, TimeUnit.MILLISECONDS); } - public void stop() { - super.stop(); - - if (zooKeeperLogPositionManager.isStart()) { - zooKeeperLogPositionManager.stop(); - } - executor.shutdownNow(); - positions.clear(); - } - - public void persistLogPosition(String destination, LogPosition logPosition) { - persistTasks.add(destination);// 添加到任务队列中进行触发 - super.persistLogPosition(destination, logPosition); - } - + @Override public LogPosition getLatestIndexBy(String destination) { - LogPosition logPostion = super.getLatestIndexBy(destination); - if (logPostion == nullPosition) { + LogPosition logPosition = memoryLogPositionManager.getLatestIndexBy(destination); + if (logPosition == nullPosition) { return null; } else { - return logPostion; + return logPosition; } } - public void setZooKeeperLogPositionManager(ZooKeeperLogPositionManager zooKeeperLogPositionManager) { - this.zooKeeperLogPositionManager = zooKeeperLogPositionManager; + @Override + public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException { + persistTasks.add(destination); + memoryLogPositionManager.persistLogPosition(destination, logPosition); } - - public void setPeriod(long period) { - this.period = period; - } - } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/ZooKeeperLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/ZooKeeperLogPositionManager.java index 996297ac..116a45e6 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/ZooKeeperLogPositionManager.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/ZooKeeperLogPositionManager.java @@ -1,33 +1,29 @@ package com.alibaba.otter.canal.parse.index; -import org.I0Itec.zkclient.exception.ZkNoNodeException; -import org.springframework.util.Assert; - -import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; import com.alibaba.otter.canal.common.utils.JsonUtils; import com.alibaba.otter.canal.common.zookeeper.ZkClientx; import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils; +import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.protocol.position.LogPosition; +import org.I0Itec.zkclient.exception.ZkNoNodeException; /** - * 基于zk的实现 - * - * @author jianghang 2012-7-7 上午10:08:27 - * @version 1.0.0 + * Created by yinxiu on 17/3/17. + * Email: marklin.hz@gmail.com */ -public class ZooKeeperLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager { +public class ZooKeeperLogPositionManager extends AbstractLogPositionManager { - private ZkClientx zkClientx; + private final ZkClientx zkClientx; - public void start() { - super.start(); - Assert.notNull(zkClientx); - } - - public void stop() { - super.stop(); + public ZooKeeperLogPositionManager(ZkClientx zkClient) { + if (zkClient == null) { + throw new NullPointerException("null zkClient"); + } + + this.zkClientx = zkClient; } + @Override public LogPosition getLatestIndexBy(String destination) { String path = ZookeeperPathUtils.getParsePath(destination); byte[] data = zkClientx.readData(path, true); @@ -38,7 +34,8 @@ public class ZooKeeperLogPositionManager extends AbstractCanalLifeCycle implemen return JsonUtils.unmarshalFromByte(data, LogPosition.class); } - public void persistLogPosition(String destination, LogPosition logPosition) { + @Override + public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException { String path = ZookeeperPathUtils.getParsePath(destination); byte[] data = JsonUtils.marshalToByte(logPosition); try { @@ -48,10 +45,4 @@ public class ZooKeeperLogPositionManager extends AbstractCanalLifeCycle implemen } } - // ================== setter / getter ================= - - public void setZkClientx(ZkClientx zkClientx) { - this.zkClientx = zkClientx; - } - } diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/group/GroupEventPaserTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/group/GroupEventPaserTest.java index b632639c..dc3de102 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/group/GroupEventPaserTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/group/GroupEventPaserTest.java @@ -2,13 +2,13 @@ package com.alibaba.otter.canal.parse.inbound.group; import java.net.InetSocketAddress; +import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager; import org.junit.Test; import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.parse.inbound.AbstractBinlogParser; import com.alibaba.otter.canal.parse.inbound.BinlogParser; import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser; -import com.alibaba.otter.canal.parse.stub.AbstractCanalLogPositionManager; import com.alibaba.otter.canal.parse.support.AuthenticationInfo; import com.alibaba.otter.canal.protocol.CanalEntry.Entry; import com.alibaba.otter.canal.protocol.position.EntryPosition; @@ -70,15 +70,16 @@ public class GroupEventPaserTest { mysqlEventPaser.setMasterPosition(defaultPosition); mysqlEventPaser.setBinlogParser(buildParser(buildAuthentication())); mysqlEventPaser.setEventSink(new EntryEventSink()); - mysqlEventPaser.setLogPositionManager(new AbstractCanalLogPositionManager() { - - public void persistLogPosition(String destination, LogPosition logPosition) { - // System.out.println(logPosition); - } - + mysqlEventPaser.setLogPositionManager(new AbstractLogPositionManager() { + @Override public LogPosition getLatestIndexBy(String destination) { return null; } + + @Override + public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException { + System.out.println(logPosition); + } }); return mysqlEventPaser; } diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogDumpTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogDumpTest.java index 5ccd816c..ffa8d11f 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogDumpTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogDumpTest.java @@ -4,11 +4,12 @@ import java.net.InetSocketAddress; import java.nio.charset.Charset; import java.util.List; +import com.alibaba.otter.canal.parse.exception.CanalParseException; +import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager; import org.junit.Assert; import org.junit.Test; import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest; -import com.alibaba.otter.canal.parse.stub.AbstractCanalLogPositionManager; import com.alibaba.otter.canal.parse.support.AuthenticationInfo; import com.alibaba.otter.canal.protocol.CanalEntry.Column; import com.alibaba.otter.canal.protocol.CanalEntry.Entry; @@ -80,17 +81,17 @@ public class LocalBinlogDumpTest { } }); - controller.setLogPositionManager(new AbstractCanalLogPositionManager() { + controller.setLogPositionManager(new AbstractLogPositionManager() { + @Override + public LogPosition getLatestIndexBy(String destination) { + return null; + } - public void persistLogPosition(String destination, LogPosition logPosition) { - System.out.println(logPosition); - } - - @Override - public LogPosition getLatestIndexBy(String destination) { - return null; - } - }); + @Override + public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException { + System.out.println(logPosition); + } + }); controller.start(); diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParserTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParserTest.java index c2badf00..c17db374 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParserTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParserTest.java @@ -7,13 +7,13 @@ import java.util.Date; import java.util.List; import java.util.concurrent.atomic.AtomicLong; +import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager; import org.junit.Assert; import org.junit.Before; import org.junit.Test; import com.alibaba.otter.canal.parse.helper.TimeoutChecker; import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest; -import com.alibaba.otter.canal.parse.stub.AbstractCanalLogPositionManager; import com.alibaba.otter.canal.parse.support.AuthenticationInfo; import com.alibaba.otter.canal.protocol.CanalEntry.Entry; import com.alibaba.otter.canal.protocol.position.EntryPosition; @@ -70,7 +70,7 @@ public class LocalBinlogEventParserTest { }); - controller.setLogPositionManager(new AbstractCanalLogPositionManager() { + controller.setLogPositionManager(new AbstractLogPositionManager() { public void persistLogPosition(String destination, LogPosition logPosition) { System.out.println(logPosition); @@ -133,7 +133,7 @@ public class LocalBinlogEventParserTest { } }); - controller.setLogPositionManager(new AbstractCanalLogPositionManager() { + controller.setLogPositionManager(new AbstractLogPositionManager() { public void persistLogPosition(String destination, LogPosition logPosition) { System.out.println(logPosition); @@ -198,7 +198,7 @@ public class LocalBinlogEventParserTest { } }); - controller.setLogPositionManager(new AbstractCanalLogPositionManager() { + controller.setLogPositionManager(new AbstractLogPositionManager() { public void persistLogPosition(String destination, LogPosition logPosition) { System.out.println(logPosition); diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlDumpTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlDumpTest.java index f4e54dfe..6c399e40 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlDumpTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlDumpTest.java @@ -4,11 +4,12 @@ import java.net.InetSocketAddress; import java.nio.charset.Charset; import java.util.List; +import com.alibaba.otter.canal.parse.exception.CanalParseException; +import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager; import org.junit.Assert; import org.junit.Test; import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest; -import com.alibaba.otter.canal.parse.stub.AbstractCanalLogPositionManager; import com.alibaba.otter.canal.parse.support.AuthenticationInfo; import com.alibaba.otter.canal.protocol.CanalEntry.Column; import com.alibaba.otter.canal.protocol.CanalEntry.Entry; @@ -83,16 +84,16 @@ public class MysqlDumpTest { } }); - controller.setLogPositionManager(new AbstractCanalLogPositionManager() { - - public void persistLogPosition(String destination, LogPosition logPosition) { - System.out.println(logPosition); - } - + controller.setLogPositionManager(new AbstractLogPositionManager() { @Override public LogPosition getLatestIndexBy(String destination) { return null; } + + @Override + public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException { + System.out.println(logPosition); + } }); controller.start(); diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParserTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParserTest.java index fbbc595c..3eaacd82 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParserTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParserTest.java @@ -5,12 +5,13 @@ import java.util.Date; import java.util.List; import java.util.concurrent.atomic.AtomicLong; +import com.alibaba.otter.canal.parse.exception.CanalParseException; +import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager; import org.junit.Assert; import org.junit.Test; import com.alibaba.otter.canal.parse.helper.TimeoutChecker; import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest; -import com.alibaba.otter.canal.parse.stub.AbstractCanalLogPositionManager; import com.alibaba.otter.canal.parse.support.AuthenticationInfo; import com.alibaba.otter.canal.protocol.CanalEntry.Entry; import com.alibaba.otter.canal.protocol.CanalEntry.EntryType; @@ -67,16 +68,16 @@ public class MysqlEventParserTest { } }); - controller.setLogPositionManager(new AbstractCanalLogPositionManager() { - - public void persistLogPosition(String destination, LogPosition logPosition) { - System.out.println(logPosition); - } - + controller.setLogPositionManager(new AbstractLogPositionManager() { @Override public LogPosition getLatestIndexBy(String destination) { return null; } + + @Override + public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException { + System.out.println(logPosition); + } }); controller.start(); @@ -136,7 +137,7 @@ public class MysqlEventParserTest { } }); - controller.setLogPositionManager(new AbstractCanalLogPositionManager() { + controller.setLogPositionManager(new AbstractLogPositionManager() { public void persistLogPosition(String destination, LogPosition logPosition) { System.out.println(logPosition); @@ -205,7 +206,7 @@ public class MysqlEventParserTest { } }); - controller.setLogPositionManager(new AbstractCanalLogPositionManager() { + controller.setLogPositionManager(new AbstractLogPositionManager() { public void persistLogPosition(String destination, LogPosition logPosition) { System.out.println(logPosition); @@ -281,7 +282,7 @@ public class MysqlEventParserTest { } }); - controller.setLogPositionManager(new AbstractCanalLogPositionManager() { + controller.setLogPositionManager(new AbstractLogPositionManager() { public void persistLogPosition(String destination, LogPosition logPosition) { System.out.println(logPosition); diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/stub/AbstractCanalLogPositionManager.java b/parse/src/test/java/com/alibaba/otter/canal/parse/stub/AbstractCanalLogPositionManager.java deleted file mode 100644 index df73735a..00000000 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/stub/AbstractCanalLogPositionManager.java +++ /dev/null @@ -1,8 +0,0 @@ -package com.alibaba.otter.canal.parse.stub; - -import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; -import com.alibaba.otter.canal.parse.index.CanalLogPositionManager; - -public abstract class AbstractCanalLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager { - -}