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

merge issue #5496 , 修复mysql rds oss订阅本地binlog过期需要再次订阅oss

Browse Source
This commit is contained in:
jianghang.loujh
2025-08-05 16:49:47 +08:00
parent 556d998db1
commit 9f2fc7409f
3 changed files with 12 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
parse/src/main/java/com/alibaba/otter/canal/parse/exception/ServerLogPurgedException.java Normal file
View File

@@ -0,0 +1 @@
package com.alibaba.otter.canal.parse.exception;

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

@@ -5,10 +5,12 @@ import java.io.InterruptedIOException;
import java.net.SocketTimeoutException; import java.net.SocketTimeoutException;
import java.nio.channels.ClosedByInterruptException; import java.nio.channels.ClosedByInterruptException;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel; 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; import com.taobao.tddl.dbsync.binlog.LogFetcher;
/** /**
@@ -99,6 +101,13 @@ public class DirectLogFetcher extends LogFetcher {
final int errno = getInt16(); final int errno = getInt16();
String sqlstate = forward(1).getFixString(SQLSTATE_LENGTH); String sqlstate = forward(1).getFixString(SQLSTATE_LENGTH);
String errmsg = getFixString(limit - position); 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 throw new IOException("Received error packet:" + " errno = " + errno + ", sqlstate = " + sqlstate
+ " errmsg = " + errmsg); + " errmsg = " + errmsg);
} else if (mark == 254) { } else if (mark == 254) {

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

@@ -6,6 +6,7 @@ import java.util.concurrent.Executors;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import com.alibaba.otter.canal.parse.exception.PositionNotFoundException; 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.ParserExceptionHandler;
import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser; import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
@@ -92,7 +93,7 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser {
} }
private void handleMysqlParserException(Throwable throwable) { 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!"); logger.info("remove rds not found position, try download rds binlog!");
executorService.execute(() -> { executorService.execute(() -> {
try { try {