From 9f2fc7409f64b0369249d160ff491332e9bbedb4 Mon Sep 17 00:00:00 2001
From: "jianghang.loujh" <jianghang.loujh@alibaba-inc.com>
Date: Tue, 5 Aug 2025 16:49:47 +0800
Subject: [PATCH] =?UTF-8?q?merge=20issue=20#5496=20,=20=E4=BF=AE=E5=A4=8Dm?=
 =?UTF-8?q?ysql=20rds=20oss=E8=AE=A2=E9=98=85=E6=9C=AC=E5=9C=B0binlog?=
 =?UTF-8?q?=E8=BF=87=E6=9C=9F=E9=9C=80=E8=A6=81=E5=86=8D=E6=AC=A1=E8=AE=A2?=
 =?UTF-8?q?=E9=98=85oss?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../canal/parse/exception/ServerLogPurgedException.java  | 1 +
 .../parse/inbound/mysql/dbsync/DirectLogFetcher.java     | 9 +++++++++
 .../inbound/mysql/rds/RdsBinlogEventParserProxy.java     | 3 ++-
 3 files changed, 12 insertions(+), 1 deletion(-)
 create mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/exception/ServerLogPurgedException.java

diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/exception/ServerLogPurgedException.java b/parse/src/main/java/com/alibaba/otter/canal/parse/exception/ServerLogPurgedException.java
new file mode 100644
index 00000000..65d79713
--- /dev/null
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/exception/ServerLogPurgedException.java
@@ -0,0 +1 @@
+package com.alibaba.otter.canal.parse.exception;

import com.alibaba.otter.canal.common.CanalException;

public class ServerLogPurgedException extends CanalException {

    public ServerLogPurgedException(String errorCode){
        super("ServerLogPurged by " + errorCode);
    }
}
\ No newline at end of file
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java
index 76cda929..18734eaa 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java
@@ -5,10 +5,12 @@ import java.io.InterruptedIOException;
 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.alibaba.otter.canal.parse.exception.ServerLogPurgedException;
 import com.taobao.tddl.dbsync.binlog.LogFetcher;
 
 /**
@@ -99,6 +101,13 @@ public class DirectLogFetcher extends LogFetcher {
                     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(
+                            " errno = " + errno + ", sqlstate = " + sqlstate + " errmsg = " + errmsg);
+                    }
+
                     throw new IOException("Received error packet:" + " errno = " + errno + ", sqlstate = " + sqlstate
                                           + " errmsg = " + errmsg);
                 } else if (mark == 254) {
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java
index c6c5312d..3fdc9c66 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java
@@ -6,6 +6,7 @@ import java.util.concurrent.Executors;
 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,7 +93,7 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser {
     }
 
     private void handleMysqlParserException(Throwable throwable) {
-        if (throwable instanceof PositionNotFoundException) {
+        if (throwable instanceof PositionNotFoundException || throwable instanceof ServerLogPurgedException) {
             logger.info("remove rds not found position, try download rds binlog!");
             executorService.execute(() -> {
                 try {