From 8df6d6fed8b9256a4591d5e0499d36a9bef3cf33 Mon Sep 17 00:00:00 2001
From: soso <sosozhuang@163.com>
Date: Tue, 28 May 2019 00:03:28 +0800
Subject: [PATCH] =?UTF-8?q?Fix:=20=E4=B8=BB=E5=A4=87=E5=88=87=E6=8D=A2?=
 =?UTF-8?q?=E5=90=8E=E6=B2=A1=E6=9C=89=E5=BE=80Kafka=E5=8F=91=E9=80=81?=
 =?UTF-8?q?=E6=95=B0=E6=8D=AE=20(#1810)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

发生主备切换时，调用canalMQStarter启停对应的destination。
---
 .../otter/canal/deployer/CanalController.java        | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
index d8ca17fd..c0692227 100644
--- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
+++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
@@ -152,6 +152,9 @@ public class CanalController {
                             try {
                                 MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
                                 embededCanalServer.start(destination);
+                                if (canalMQStarter != null) {
+                                    canalMQStarter.startDestination(destination);
+                                }
                             } finally {
                                 MDC.remove(CanalConstants.MDC_DESTINATION);
                             }
@@ -160,6 +163,9 @@ public class CanalController {
                         public void processActiveExit() {
                             try {
                                 MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
+                                if (canalMQStarter != null) {
+                                    canalMQStarter.stopDestination(destination);
+                                }
                                 embededCanalServer.stop(destination);
                             } finally {
                                 MDC.remove(CanalConstants.MDC_DESTINATION);
@@ -234,9 +240,6 @@ public class CanalController {
                         ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
                         if (!config.getLazy() && !runningMonitor.isStart()) {
                             runningMonitor.start();
-                            if (canalMQStarter != null) {
-                                canalMQStarter.startDestination(destination);
-                            }
                         }
                     }
                 }
@@ -245,9 +248,6 @@ public class CanalController {
                     // 此处的stop，代表强制退出，非HA机制，所以需要退出HA的monitor和配置信息
                     InstanceConfig config = instanceConfigs.remove(destination);
                     if (config != null) {
-                        if (canalMQStarter != null) {
-                            canalMQStarter.stopDestination(destination);
-                        }
                         embededCanalServer.stop(destination);
                         ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
                         if (runningMonitor.isStart()) {