diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java index 577e354a..3e15db30 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java @@ -64,6 +64,27 @@ public class SyncSwitch { } } + public synchronized void refresh() { + for (String destination : adapterCanalConfig.DESTINATIONS) { + BooleanMutex booleanMutex; + if (mode == Mode.DISTRIBUTED) { + CuratorFramework curator = curatorClient.getCurator(); + booleanMutex = DISTRIBUTED_LOCK.get(destination); + if (booleanMutex == null) { + BooleanMutex mutex = new BooleanMutex(true); + initMutex(curator, destination, mutex); + DISTRIBUTED_LOCK.put(destination, mutex); + startListen(destination, mutex); + } + } else { + booleanMutex = LOCAL_LOCK.get(destination); + if (booleanMutex == null) { + LOCAL_LOCK.put(destination, new BooleanMutex(true)); + } + } + } + } + @SuppressWarnings("resource") private synchronized void startListen(String destination, BooleanMutex mutex) { try { diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java index 48cad990..58d1a513 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java @@ -53,6 +53,8 @@ public class CanalAdapterService { return; } try { + syncSwitch.refresh(); + logger.info("## syncSwitch refreshed."); logger.info("## start the canal client adapters."); adapterLoader = new CanalAdapterLoader(adapterCanalConfig); adapterLoader.init();