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

Fix: 主备切换后没有往Kafka发送数据 (#1810)

Browse Source 
发生主备切换时，调用canalMQStarter启停对应的destination。
This commit is contained in:
soso
2019-05-28 00:03:28 +08:00
committed by agapple
parent f9989e0856
commit 8df6d6fed8
1 changed files with 6 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
View File

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