From 50a570290a30348ea5005eacb7345ee85032d8cc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sunxien=C2=AE?= <81640257+sunxien@users.noreply.github.com> Date: Fri, 16 May 2025 15:12:42 +0800 Subject: [PATCH] =?UTF-8?q?fix(canal/client-adapter):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?adapter=E6=8F=92=E4=BB=B6=E9=87=8D=E8=AF=95=E9=94=99=E8=AF=AF?= =?UTF-8?q?=E7=BC=BA=E9=99=B7=20(#5427)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(canal/prometheus): 修复延迟指标的缺陷 -【缺陷描述】 Admin控制台修改canal.properties配置后，Server自动进行重启，在Grafan上观测到同步实例的PUT/GET/ACK延迟居高不下。 -【原因定位】 源库心跳正常触发，MemoryStoreWithBuffer正常推进位点，profiling 正常统计，但是延迟指标依然是越来越来越高。Prometheus调用collect接口采集到的exec time时间始终是固定不变的。经debug排查到StoreCollector采集器内存hold的StoreMetricsHolder与CanalInstance实例中的引用已经不同啦，CanalIntance重启时已经被重建过一份新的实例，但是StoreMetricsHolder却没有保存到内存Hold中。原因是Map.putIfAbsent调用引起。 -【修复效果】 - 修复后，重启Server心跳正常推进，延迟瞬间降下来。梳理其他Collector代码都是调用Map.put，只有这里使用putIfAbsent可能是粗心导致吧。修复效果 * fix(canal/client-adapter): 修复adapter插件重试错误缺陷 -【缺陷描述】 adapter运行期间，针对server或者Instance进行过一次重启或者断网测试，发现adapter通过client拉取数据报错以后持续打印错误日志，永远不得恢复。虽然adapter processor 针对错误有重试逻辑，但是client的tcp连接已经发生中断，且无法恢复。 -【问题定位】 adapter processor 针对错误的处理比较粗，需要细分是从上游get message报错，还是下游sync data报错？ 如果是 get message报错，特别是tcp 断开，需要跳出重试，重新尝试重连。如果是下游 sync data 报错，再考虑重试写入。下游插件开发者自行保证连接重连机制、写入幂等。框架只提供重试。 - 说明：顺带更新一下 .gitignore 文件 --- .gitignore | 1 + .../adapter/launcher/loader/AdapterProcessor.java | 15 ++++++++++++++- 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index f100e606..c94054d4 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,7 @@ test-output/ .settings/ tmp temp +logs/ *.log antx.properties otter.properties diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AdapterProcessor.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AdapterProcessor.java index e8ca4e9b..8dd5e168 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AdapterProcessor.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AdapterProcessor.java @@ -8,6 +8,8 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import com.alibaba.otter.canal.protocol.exception.CanalClientException; +import org.apache.commons.lang.exception.ExceptionUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -186,7 +188,7 @@ public class AdapterProcessor { logger.info("=============> Start to connect destination: {} <=============", this.canalDestination); canalMsgConsumer.connect(); logger.info("=============> Subscribe destination: {} succeed <=============", this.canalDestination); - while (running) { + out: while (running) { try { syncSwitch.get(canalDestination, 1L, TimeUnit.MINUTES); } catch (TimeoutException e) { @@ -216,6 +218,17 @@ public class AdapterProcessor { } break; } catch (Exception e) { + Throwable th = e.getCause(); + // Handle source error when getting message + if (th instanceof CanalClientException) { + String message = ExceptionUtils.getRootCauseMessage(th); + if (message.contains("end of stream when reading header") + || message.contains("Connection reset by peer") || message.contains("Broken pipe")) { + logger.error("Sync failed, reconnect to canal instance. Error: {}", message); + break out; + } + } + // Handle sink error if (i != retry - 1) { canalMsgConsumer.rollback(); // 处理失败, 回滚数据 logger.error(e.getMessage() + " Error sync and rollback, execute times: " + (i + 1));