diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java index f8438ae4..333b0805 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java @@ -4,6 +4,7 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.Future; import org.slf4j.Logger; @@ -36,7 +37,9 @@ public abstract class AbstractCanalAdapterWorker { protected SyncSwitch syncSwitch; - public AbstractCanalAdapterWorker(){ + public AbstractCanalAdapterWorker(List> canalOuterAdapters){ + this.canalOuterAdapters = canalOuterAdapters; + this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); syncSwitch = (SyncSwitch) SpringContext.getBean(SyncSwitch.class); } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java index 1ff58e9a..e74641ca 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java @@ -26,7 +26,7 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { public CanalAdapterKafkaWorker(String bootstrapServers, String topic, String groupId, List> canalOuterAdapters, boolean flatMessage){ - this.canalOuterAdapters = canalOuterAdapters; + super(canalOuterAdapters); this.topic = topic; this.canalDestination = topic; this.flatMessage = flatMessage; diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java index fce63946..db80d602 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java @@ -24,12 +24,12 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker { public CanalAdapterRocketMQWorker(String nameServers, String topic, String groupId, List> canalOuterAdapters, boolean flatMessage){ - logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId); - this.canalOuterAdapters = canalOuterAdapters; + super(canalOuterAdapters); this.topic = topic; this.flatMessage = flatMessage; this.canalDestination = topic; this.connector = RocketMQCanalConnectors.newRocketMQConnector(nameServers, topic, groupId, flatMessage); + logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId); } @Override diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java index 2a6dc1f7..826e9ccd 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java @@ -34,7 +34,7 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { */ public CanalAdapterWorker(String canalDestination, SocketAddress address, List> canalOuterAdapters){ - this.canalOuterAdapters = canalOuterAdapters; + super(canalOuterAdapters); this.canalDestination = canalDestination; groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); connector = CanalConnectors.newSingleConnector(address, canalDestination, "", ""); @@ -49,7 +49,7 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { */ public CanalAdapterWorker(String canalDestination, String zookeeperHosts, List> canalOuterAdapters){ - this.canalOuterAdapters = canalOuterAdapters; + super(canalOuterAdapters); this.canalDestination = canalDestination; groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); connector = CanalConnectors.newClusterConnector(zookeeperHosts, canalDestination, "", "");