diff --git a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/monitor/ESConfigMonitor.java b/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/monitor/ESConfigMonitor.java index 527d292b..824072e3 100644 --- a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/monitor/ESConfigMonitor.java +++ b/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/monitor/ESConfigMonitor.java @@ -26,11 +26,11 @@ import com.alibaba.otter.canal.client.adapter.support.Util; public class ESConfigMonitor { - private static final Logger logger = LoggerFactory.getLogger(ESConfigMonitor.class); + private static final Logger logger = LoggerFactory.getLogger(ESConfigMonitor.class); - private static final String adapterName = "es"; + private String adapterName; - private ESAdapter esAdapter; + private ESAdapter esAdapter; private Properties envProperties; @@ -39,6 +39,7 @@ public class ESConfigMonitor { public void init(ESAdapter esAdapter, Properties envProperties) { this.esAdapter = esAdapter; this.envProperties = envProperties; + this.adapterName = envProperties.getProperty("es.version"); File confDir = Util.getConfDirPath(adapterName); try { FileAlterationObserver observer = new FileAlterationObserver(confDir, @@ -130,7 +131,7 @@ public class ESConfigMonitor { private void addConfigToCache(File file, ESSyncConfig config) { esAdapter.getEsSyncConfig().put(file.getName(), config); - esAdapter.addSyncConfigToCache(file.getName(),config); + esAdapter.addSyncConfigToCache(file.getName(), config); } private void deleteConfigFromCache(File file) { diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java index 51e0c3dd..3217af91 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java @@ -7,6 +7,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; @@ -63,10 +64,10 @@ public class CanalAdapterLoader { if ("tcp".equalsIgnoreCase(canalClientConfig.getMode())) { // 初始化canal-client的适配器 for (CanalClientConfig.CanalAdapter canalAdapter : canalClientConfig.getCanalAdapters()) { - List> canalOuterAdapterGroups = new ArrayList<>(); + List> canalOuterAdapterGroups = new CopyOnWriteArrayList<>(); for (CanalClientConfig.Group connectorGroup : canalAdapter.getGroups()) { - List canalOutConnectors = new ArrayList<>(); + List canalOutConnectors = new CopyOnWriteArrayList<>(); for (OuterAdapterConfig c : connectorGroup.getOuterAdapters()) { loadAdapter(c, canalOutConnectors); } @@ -94,8 +95,8 @@ public class CanalAdapterLoader { // 初始化canal-client-kafka的适配器 for (CanalClientConfig.CanalAdapter canalAdapter : canalClientConfig.getCanalAdapters()) { for (CanalClientConfig.Group group : canalAdapter.getGroups()) { - List> canalOuterAdapterGroups = new ArrayList<>(); - List canalOuterAdapters = new ArrayList<>(); + List> canalOuterAdapterGroups = new CopyOnWriteArrayList<>(); + List canalOuterAdapters = new CopyOnWriteArrayList<>(); for (OuterAdapterConfig config : group.getOuterAdapters()) { loadAdapter(config, canalOuterAdapters); } @@ -117,8 +118,8 @@ public class CanalAdapterLoader { // 初始化canal-client-rocketMQ的适配器 for (CanalClientConfig.CanalAdapter canalAdapter : canalClientConfig.getCanalAdapters()) { for (CanalClientConfig.Group group : canalAdapter.getGroups()) { - List> canalOuterAdapterGroups = new ArrayList<>(); - List canalOuterAdapters = new ArrayList<>(); + List> canalOuterAdapterGroups = new CopyOnWriteArrayList<>(); + List canalOuterAdapters = new CopyOnWriteArrayList<>(); for (OuterAdapterConfig config : group.getOuterAdapters()) { loadAdapter(config, canalOuterAdapters); } @@ -146,8 +147,8 @@ public class CanalAdapterLoader { // 初始化canal-client-rabbitMQ的适配器 for (CanalClientConfig.CanalAdapter canalAdapter : canalClientConfig.getCanalAdapters()) { for (CanalClientConfig.Group group : canalAdapter.getGroups()) { - List> canalOuterAdapterGroups = new ArrayList<>(); - List canalOuterAdapters = new ArrayList<>(); + List> canalOuterAdapterGroups = new CopyOnWriteArrayList<>(); + List canalOuterAdapters = new CopyOnWriteArrayList<>(); for (OuterAdapterConfig config : group.getOuterAdapters()) { loadAdapter(config, canalOuterAdapters); } @@ -184,7 +185,7 @@ public class CanalAdapterLoader { if (propertySource instanceof EnumerablePropertySource) { String[] names = ((EnumerablePropertySource) propertySource).getPropertyNames(); for (String name : names) { - Object val = propertySource.getProperty(name); + Object val = env.getProperty(name); if (val != null) { evnProperties.put(name, val); }