From 34e9440383068c28c24a215e39b55fe790117f0d Mon Sep 17 00:00:00 2001 From: "jianghang.loujh" Date: Mon, 4 Sep 2023 19:34:06 +0800 Subject: [PATCH] fixed issue #4864 , connector.disconnect optimize --- .../tcp/consumer/CanalTCPConsumer.java | 20 ++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/connector/tcp-connector/src/main/java/com/alibaba/otter/canal/connector/tcp/consumer/CanalTCPConsumer.java b/connector/tcp-connector/src/main/java/com/alibaba/otter/canal/connector/tcp/consumer/CanalTCPConsumer.java index bc7966bf..a773363a 100644 --- a/connector/tcp-connector/src/main/java/com/alibaba/otter/canal/connector/tcp/consumer/CanalTCPConsumer.java +++ b/connector/tcp-connector/src/main/java/com/alibaba/otter/canal/connector/tcp/consumer/CanalTCPConsumer.java @@ -1,5 +1,13 @@ package com.alibaba.otter.canal.connector.tcp.consumer; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.util.List; +import java.util.Properties; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.lang.StringUtils; + import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.impl.ClusterCanalConnector; import com.alibaba.otter.canal.client.impl.ClusterNodeAccessStrategy; @@ -12,14 +20,6 @@ import com.alibaba.otter.canal.connector.core.util.MessageUtil; import com.alibaba.otter.canal.connector.tcp.config.TCPConstants; import com.alibaba.otter.canal.protocol.Message; -import org.apache.commons.lang.StringUtils; - -import java.net.InetSocketAddress; -import java.net.SocketAddress; -import java.util.List; -import java.util.Properties; -import java.util.concurrent.TimeUnit; - /** * TCP 消费者连接器, 一个destination对应一个SPI实例 * @@ -99,7 +99,9 @@ public class CanalTCPConsumer implements CanalMsgConsumer { @Override public void disconnect() { - canalConnector.unsubscribe(); + // tcp模式下，因为是单tcp消费，避免adapter异常断开时直接unsubscribe + // unsubscribe发送给canal-server会导致清理cursor位点,如果此时canal-server出现重启,就会丢失binlog数据 + // canalConnector.unsubscribe(); canalConnector.disconnect(); } }