修复mysql rds oss订阅本地binlog过期需要再次订阅oss (#5496)

* fix: fix rds oss server binlog expired

* fix: fix rds oss server binlog expired

* fix: fix rds oss server binlog expired

---------

Co-authored-by: 子茗 <[email protected]>

Author: ziming-ai

parse/src/main/java/com/alibaba/otter/canal/parse/exception/ServerLogPurgedException.java

@@ -0,0 +1,8 @@
+package com.alibaba.otter.canal.parse.exception;
+import com.alibaba.otter.canal.common.CanalException;
+
+public class ServerLogPurgedException extends CanalException {
+    public ServerLogPurgedException(String errorCode) {
+        super("ServerLogPurged0: " + errorCode);
+    }
+}

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

@@ -5,11 +5,13 @@
 import java.net.SocketTimeoutException;
 import java.nio.channels.ClosedByInterruptException;
 
+import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel;
 import com.taobao.tddl.dbsync.binlog.LogFetcher;
+import com.alibaba.otter.canal.parse.exception.ServerLogPurgedException;
 
 /**
  * 基于socket的logEvent实现
@@ -99,6 +101,11 @@ public boolean fetch() throws IOException {
                     final int errno = getInt16();
                     String sqlstate = forward(1).getFixString(SQLSTATE_LENGTH);
                     String errmsg = getFixString(limit - position);
+                    if (StringUtils.containsIgnoreCase(errmsg, "not find first log file name")
+                            || StringUtils.containsIgnoreCase(errmsg, "purged binary logs")) {
+                        // 开始 dump 后，server 位点过期，DUMP 和 DUMP_GTID 两种错误信息
+                        throw new ServerLogPurgedException(errmsg);
+                    }
                     throw new IOException("Received error packet:" + " errno = " + errno + ", sqlstate = " + sqlstate
                                           + " errmsg = " + errmsg);
                 } else if (mark == 254) {

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

@@ -6,6 +6,7 @@
 import org.apache.commons.lang.StringUtils;
 
 import com.alibaba.otter.canal.parse.exception.PositionNotFoundException;
+import com.alibaba.otter.canal.parse.exception.ServerLogPurgedException;
 import com.alibaba.otter.canal.parse.inbound.ParserExceptionHandler;
 import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
 
@@ -92,8 +93,10 @@ public void start() {
     }
 
     private void handleMysqlParserException(Throwable throwable) {
-        if (throwable instanceof PositionNotFoundException) {
-            logger.info("remove rds not found position, try download rds binlog!");
+        if (throwable instanceof PositionNotFoundException
+                || throwable instanceof ServerLogPurgedException) {
+            logger.info("remove rds not found position, try download rds binlog! {} : {}",
+                    throwable.getClass().getSimpleName(), throwable.getMessage());
             executorService.execute(() -> {
                 try {
                     logger.info("stop mysql parser!");