chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

修复最近位置被remove掉，导致重复消息的问题 (#3276)

Browse Source
This commit is contained in:
binbin
2021-04-16 22:03:51 +08:00
committed by GitHub
parent b1dc3821ea
commit f66baeabf1
2 changed files with 4 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
meta/src/main/java/com/alibaba/otter/canal/meta/FileMixedMetaManager.java
View File

@@ -91,6 +91,8 @@ public class FileMixedMetaManager extends MemoryMetaManager implements CanalMeta
for (ClientIdentity clientIdentity : tasks) { for (ClientIdentity clientIdentity : tasks) {
MDC.put("destination", String.valueOf(clientIdentity.getDestination())); MDC.put("destination", String.valueOf(clientIdentity.getDestination()));
try { try {
updateCursorTasks.remove(clientIdentity);
// 定时将内存中的最新值刷到file中，多次变更只刷一次 // 定时将内存中的最新值刷到file中，多次变更只刷一次
if (logger.isInfoEnabled()) { if (logger.isInfoEnabled()) {
LogPosition cursor = (LogPosition) getCursor(clientIdentity); LogPosition cursor = (LogPosition) getCursor(clientIdentity);
@@ -100,7 +102,6 @@ public class FileMixedMetaManager extends MemoryMetaManager implements CanalMeta
cursor.getIdentity().getSourceAddress().toString()); cursor.getIdentity().getSourceAddress().toString());
} }
flushDataToFile(clientIdentity.getDestination()); flushDataToFile(clientIdentity.getDestination());
updateCursorTasks.remove(clientIdentity);
} catch (Throwable e) { } catch (Throwable e) {
// ignore // ignore
logger.error("period update" + clientIdentity.toString() + " curosr failed!", e); logger.error("period update" + clientIdentity.toString() + " curosr failed!", e);

3
meta/src/main/java/com/alibaba/otter/canal/meta/PeriodMixedMetaManager.java
View File

@@ -79,9 +79,10 @@ public class PeriodMixedMetaManager extends MemoryMetaManager implements CanalMe
List<ClientIdentity> tasks = new ArrayList<>(updateCursorTasks); List<ClientIdentity> tasks = new ArrayList<>(updateCursorTasks);
for (ClientIdentity clientIdentity : tasks) { for (ClientIdentity clientIdentity : tasks) {
try { try {
updateCursorTasks.remove(clientIdentity);
// 定时将内存中的最新值刷到zookeeper中，多次变更只刷一次 // 定时将内存中的最新值刷到zookeeper中，多次变更只刷一次
zooKeeperMetaManager.updateCursor(clientIdentity, getCursor(clientIdentity)); zooKeeperMetaManager.updateCursor(clientIdentity, getCursor(clientIdentity));
updateCursorTasks.remove(clientIdentity);
} catch (Throwable e) { } catch (Throwable e) {
// ignore // ignore
logger.error("period update" + clientIdentity.toString() + " curosr failed!", e); logger.error("period update" + clientIdentity.toString() + " curosr failed!", e);