From be0d945d7d221c15b8191f3da5eadfc0242fe613 Mon Sep 17 00:00:00 2001
From: lwd-coding <654547217@qq.com>
Date: Fri, 13 Sep 2024 17:02:19 +0800
Subject: [PATCH] =?UTF-8?q?=E5=BD=93zk=E5=87=BA=E7=8E=B0session=20time=20o?=
 =?UTF-8?q?ut=E5=8F=AF=E8=83=BD=E4=BC=9A=E5=AF=BC=E8=87=B4instance?=
 =?UTF-8?q?=E5=87=BA=E7=8E=B0=E5=A4=9A=E4=B8=AA=E6=B6=88=E8=B4=B9=E8=80=85?=
 =?UTF-8?q?=EF=BC=8C=E4=BB=8E=E8=80=8C=E5=87=BA=E7=8E=B0=E4=B9=B1=E5=BA=8F?=
 =?UTF-8?q?=20(#5270)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../java/com/alibaba/otter/canal/server/CanalMQStarter.java | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java
index 58f7cd66..d64a14c4 100644
--- a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java
+++ b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java
@@ -15,6 +15,7 @@ import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;
 
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -106,7 +107,10 @@ public class CanalMQStarter {
 
     public synchronized void startDestination(String destination) {
         CanalInstance canalInstance = canalServer.getCanalInstances().get(destination);
-        if (canalInstance != null) {
+        CanalMQRunnable canalMQRunable = canalMQWorks.get(destination);
+        //当zk出现session time out可能会导致instance出现多个消费者，从而出现乱序
+        //因此当抢占成功的实例已经启动CanalMQRunnable时就不需要stop然后start了
+        if (canalInstance != null && Objects.isNull(canalMQRunable)) {
             stopDestination(destination);
             CanalMQRunnable canalMQRunnable = new CanalMQRunnable(destination);
             canalMQWorks.put(canalInstance.getDestination(), canalMQRunnable);