在位点出现问题的时候，清空位点缓存，这样只要修正配置文件就可以获取最新，而不需要重启canal

meta/src/main/java/com/alibaba/otter/canal/meta/CanalMetaManager.java

@@ -90,4 +90,11 @@ void addBatch(ClientIdentity clientIdentity, PositionRange positionRange, Long b
      */
     void clearAllBatchs(ClientIdentity clientIdentity) throws CanalMetaManagerException;
 
+    /**
+     * 清楚游标緩存
+     * @param clientIdentity
+     * @throws CanalMetaManagerException
+     */
+   default void removeCursorsCache(ClientIdentity clientIdentity) throws CanalMetaManagerException {};
+
 }

meta/src/main/java/com/alibaba/otter/canal/meta/MemoryMetaManager.java

@@ -113,6 +113,11 @@ public void clearAllBatchs(ClientIdentity clientIdentity) throws CanalMetaManage
         batches.get(clientIdentity).clearPositionRanges();
     }
 
+    @Override
+    public void removeCursorsCache(ClientIdentity clientIdentity) throws CanalMetaManagerException {
+        cursors.remove(clientIdentity);
+    }
+
     // ============================
 
     public static class MemoryClientIdentityBatch {

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java

@@ -327,6 +327,7 @@ public boolean sink(EVENT event) {
                     eventSink.interrupt();
                     transactionBuffer.reset();// 重置一下缓冲队列，重新记录数据
                     binlogParser.reset();// 重新置位
+                    removeLogPositionCacheWhenError(destination);
                     if (multiStageCoprocessor != null && multiStageCoprocessor.isStart()) {
                         // 处理 RejectedExecutionException
                         try {
@@ -355,6 +356,8 @@ public boolean sink(EVENT event) {
         parseThread.start();
     }
 
+
+
     public void stop() {
         super.stop();
 
@@ -471,6 +474,14 @@ protected void processDumpError(Throwable e) {
         // do nothing
     }
 
+    /**
+     * 如果是缓存，则清理缓存,允许修改底层的持久化文件的时候无需重启canal
+     * @param destination
+     */
+    protected void removeLogPositionCacheWhenError(String destination) {
+        // do nothing
+    }
+
     protected void startHeartBeat(ErosaConnection connection) {
         lastEntryTime = 0L; // 初始化
         if (timer == null) {// lazy初始化一下

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java

@@ -859,6 +859,14 @@ protected void processDumpError(Throwable e) {
         super.processDumpError(e);
     }
 
+    @Override
+    protected void removeLogPositionCacheWhenError(String destination) {
+        if (dumpErrorCount > dumpErrorCountThreshold + 1) {
+            // 这里+1 是允许主从切换的时候重试一下再清缓存
+            logPositionManager.removeLogPositionCache(destination);
+        }
+    }
+
     public void setSupportBinlogFormats(String formatStrs) {
         String[] formats = StringUtils.split(formatStrs, ',');
         if (formats != null) {

parse/src/main/java/com/alibaba/otter/canal/parse/index/CanalLogPositionManager.java

@@ -13,4 +13,9 @@ public interface CanalLogPositionManager extends CanalLifeCycle {
 
     void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException;
 
+    /**
+     * 一般使用在异常场景清理缓存，这样持久化的存储更新后能够得到最新的数据，不需要重启机器
+     */
+   default void removeLogPositionCache(String destination){}
+
 }

parse/src/main/java/com/alibaba/otter/canal/parse/index/FailbackLogPositionManager.java

@@ -80,4 +80,10 @@ public void persistLogPosition(String destination, LogPosition logPosition) thro
             secondary.persistLogPosition(destination, logPosition);
         }
     }
+
+    @Override
+    public void removeLogPositionCache(String destination) {
+        primary.removeLogPositionCache(destination);
+        secondary.removeLogPositionCache(destination);
+    }
 }

parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java

@@ -188,4 +188,9 @@ private LogPosition loadDataFromFile(File dataFile) {
             throw new CanalMetaManagerException(e);
         }
     }
+
+    @Override
+    public void removeLogPositionCache(String destination) {
+        memoryLogPositionManager.removeLogPositionCache(destination);
+    }
 }

parse/src/main/java/com/alibaba/otter/canal/parse/index/MemoryLogPositionManager.java

@@ -40,4 +40,8 @@ public Set<String> destinations() {
         return positions.keySet();
     }
 
+    @Override
+    public void removeLogPositionCache(String destination) {
+        positions.remove(destination);
+    }
 }

parse/src/main/java/com/alibaba/otter/canal/parse/index/MetaLogPositionManager.java

@@ -75,4 +75,17 @@ public void persistLogPosition(String destination, LogPosition logPosition) thro
         // do nothing
         logger.info("destination [{}] persist LogPosition:{}", destination, logPosition);
     }
+
+    @Override
+    public void removeLogPositionCache(String destination) {
+        List<ClientIdentity> clientIdentities = metaManager.listAllSubscribeInfo(destination);
+        LogPosition result = null;
+        if (CollectionUtils.isEmpty(clientIdentities)) {
+           return;
+        }
+        for (ClientIdentity clientIdentity : clientIdentities) {
+            metaManager.removeCursorsCache(clientIdentity);
+        }
+
+    }
 }

parse/src/main/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManager.java

@@ -81,4 +81,9 @@ public void persistLogPosition(final String destination, final LogPosition logPo
             }
         });
     }
+
+    @Override
+    public void removeLogPositionCache(String destination) {
+        memoryLogPositionManager.removeLogPositionCache(destination);
+    }
 }

parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java

@@ -112,4 +112,9 @@ public void persistLogPosition(String destination, LogPosition logPosition) thro
         persistTasks.add(destination);
         memoryLogPositionManager.persistLogPosition(destination, logPosition);
     }
+
+    @Override
+    public void removeLogPositionCache(String destination) {
+        memoryLogPositionManager.removeLogPositionCache(destination);
+    }
 }