repair issue #3401 (#3923)

* 解决OutAdapter单例导致的问题，如 1.多个es7 outAdapter的多线程问题（共享同一个es bulk request），2.多个es7 outAdapter共用一份es配置问题

* 1. 优化adapter和监听器之间的代码结构
2. 监听器只处理匹配的配置文件
3. bugfix-es adapter配置文件回调时漏写esVersion
理论上一个adapter实例对应一个config配置文件

Co-authored-by: agapple <jianghang.loujh@alibaba-inc.com>

client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java

@@ -1,9 +1,25 @@
 package com.alibaba.otter.canal.client.adapter.hbase;
 
+import com.alibaba.otter.canal.client.adapter.OuterAdapter;
+import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig;
+import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfigLoader;
+import com.alibaba.otter.canal.client.adapter.hbase.monitor.HbaseConfigMonitor;
+import com.alibaba.otter.canal.client.adapter.hbase.service.HbaseEtlService;
+import com.alibaba.otter.canal.client.adapter.hbase.service.HbaseSyncService;
+import com.alibaba.otter.canal.client.adapter.hbase.support.HbaseTemplate;
+import com.alibaba.otter.canal.client.adapter.support.Dml;
+import com.alibaba.otter.canal.client.adapter.support.EtlResult;
+import com.alibaba.otter.canal.client.adapter.support.FileName2KeyMapping;
+import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
+import com.alibaba.otter.canal.client.adapter.support.SPI;
+import com.alibaba.otter.canal.client.adapter.support.Util;
 import java.io.IOException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
 import java.util.concurrent.ConcurrentHashMap;
-
 import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HBaseConfiguration;
@@ -16,18 +32,6 @@ import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.alibaba.otter.canal.client.adapter.OuterAdapter;
-import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig;
-import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfigLoader;
-import com.alibaba.otter.canal.client.adapter.hbase.monitor.HbaseConfigMonitor;
-import com.alibaba.otter.canal.client.adapter.hbase.service.HbaseEtlService;
-import com.alibaba.otter.canal.client.adapter.hbase.service.HbaseSyncService;
-import com.alibaba.otter.canal.client.adapter.hbase.support.HbaseTemplate;
-import com.alibaba.otter.canal.client.adapter.support.Dml;
-import com.alibaba.otter.canal.client.adapter.support.EtlResult;
-import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
-import com.alibaba.otter.canal.client.adapter.support.SPI;
-
 /**
  * HBase外部适配器
  *
@@ -49,6 +53,8 @@ public class HbaseAdapter implements OuterAdapter {
 
     private Properties                              envProperties;
 
+    private OuterAdapterConfig                      configuration;
+
     public Map<String, MappingConfig> getHbaseMapping() {
         return hbaseMapping;
     }
@@ -61,33 +67,12 @@ public class HbaseAdapter implements OuterAdapter {
     public void init(OuterAdapterConfig configuration, Properties envProperties) {
         try {
             this.envProperties = envProperties;
+            this.configuration = configuration;
             Map<String, MappingConfig> hbaseMappingTmp = MappingConfigLoader.load(envProperties);
             // 过滤不匹配的key的配置
-            hbaseMappingTmp.forEach((key, mappingConfig) -> {
-                if ((mappingConfig.getOuterAdapterKey() == null && configuration.getKey() == null)
-                    || (mappingConfig.getOuterAdapterKey() != null
-                        && mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) {
-                    hbaseMapping.put(key, mappingConfig);
-                }
+            hbaseMappingTmp.forEach((key, config) -> {
+                addConfig(key, config);
             });
-            for (Map.Entry<String, MappingConfig> entry : hbaseMapping.entrySet()) {
-                String configName = entry.getKey();
-                MappingConfig mappingConfig = entry.getValue();
-                String k;
-                if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
-                    k = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "-"
-                        + StringUtils.trimToEmpty(mappingConfig.getGroupId()) + "_"
-                        + mappingConfig.getHbaseMapping().getDatabase() + "-"
-                        + mappingConfig.getHbaseMapping().getTable();
-                } else {
-                    k = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "_"
-                        + mappingConfig.getHbaseMapping().getDatabase() + "-"
-                        + mappingConfig.getHbaseMapping().getTable();
-                }
-                Map<String, MappingConfig> configMap = mappingConfigCache.computeIfAbsent(k,
-                    k1 -> new ConcurrentHashMap<>());
-                configMap.put(configName, mappingConfig);
-            }
 
             Map<String, String> properties = configuration.getProperties();
 
@@ -223,4 +208,62 @@ public class HbaseAdapter implements OuterAdapter {
         }
         return null;
     }
+
+    private void addSyncConfigToCache(String configName, MappingConfig mappingConfig) {
+        String k;
+        if (envProperties != null && !"tcp"
+                .equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
+            k = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "-" + StringUtils
+                    .trimToEmpty(mappingConfig.getGroupId()) + "_" + mappingConfig.getHbaseMapping()
+                    .getDatabase() + "-" + mappingConfig.getHbaseMapping().getTable();
+        } else {
+            k = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "_" + mappingConfig
+                    .getHbaseMapping().getDatabase() + "-" + mappingConfig.getHbaseMapping()
+                    .getTable();
+        }
+        Map<String, MappingConfig> configMap = mappingConfigCache
+                .computeIfAbsent(k, k1 -> new ConcurrentHashMap<>());
+        configMap.put(configName, mappingConfig);
+    }
+
+    public boolean addConfig(String fileName, MappingConfig config) {
+        if (match(config)) {
+            hbaseMapping.put(fileName, config);
+            addSyncConfigToCache(fileName, config);
+            FileName2KeyMapping.register(getClass().getAnnotation(SPI.class).value(), fileName,
+                    configuration.getKey());
+            return true;
+        }
+        return false;
+    }
+
+    public void updateConfig(String fileName, MappingConfig config) {
+        if (config.getOuterAdapterKey() != null && !config.getOuterAdapterKey()
+                .equals(configuration.getKey())) {
+            // 理论上不允许改这个 因为本身就是通过这个关联起Adapter和Config的
+            throw new RuntimeException("not allow to change outAdapterKey");
+        }
+        hbaseMapping.put(fileName, config);
+        addSyncConfigToCache(fileName, config);
+    }
+
+    public void deleteConfig(String fileName) {
+        hbaseMapping.remove(fileName);
+        for (Map<String, MappingConfig> configMap : mappingConfigCache.values()) {
+            if (configMap != null) {
+                configMap.remove(fileName);
+            }
+        }
+        FileName2KeyMapping.unregister(getClass().getAnnotation(SPI.class).value(), fileName);
+    }
+
+    private boolean match(MappingConfig config) {
+        boolean sameMatch = config.getOuterAdapterKey() != null && config.getOuterAdapterKey()
+                .equalsIgnoreCase(configuration.getKey());
+        boolean prefixMatch = config.getOuterAdapterKey() == null && configuration.getKey()
+                .startsWith(StringUtils
+                        .join(new String[]{Util.AUTO_GENERATED_PREFIX, config.getDestination(),
+                                config.getGroupId()}, '-'));
+        return sameMatch || prefixMatch;
+    }
 }

client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/monitor/HbaseConfigMonitor.java

@@ -5,20 +5,15 @@ import com.alibaba.otter.canal.client.adapter.hbase.HbaseAdapter;
 import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig;
 import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
 import com.alibaba.otter.canal.client.adapter.support.Util;
-
+import java.io.File;
+import java.util.Properties;
 import org.apache.commons.io.filefilter.FileFilterUtils;
 import org.apache.commons.io.monitor.FileAlterationListenerAdaptor;
 import org.apache.commons.io.monitor.FileAlterationMonitor;
 import org.apache.commons.io.monitor.FileAlterationObserver;
-import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.File;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Properties;
-
 public class HbaseConfigMonitor {
 
     private static final Logger   logger      = LoggerFactory.getLogger(HbaseConfigMonitor.class);
@@ -37,7 +32,7 @@ public class HbaseConfigMonitor {
         File confDir = Util.getConfDirPath(adapterName);
         try {
             FileAlterationObserver observer = new FileAlterationObserver(confDir,
-                FileFilterUtils.and(FileFilterUtils.fileFileFilter(), FileFilterUtils.suffixFileFilter("yml")));
+                    FileFilterUtils.and(FileFilterUtils.fileFileFilter(), FileFilterUtils.suffixFileFilter("yml")));
             FileListener listener = new FileListener();
             observer.addListener(listener);
             fileMonitor = new FileAlterationMonitor(3000, observer);
@@ -70,9 +65,11 @@ public class HbaseConfigMonitor {
                     return;
                 }
                 config.validate();
-                addConfigToCache(file, config);
-
-                logger.info("Add a new hbase mapping config: {} to canal adapter", file.getName());
+                boolean result = hbaseAdapter.addConfig(file.getName(), config);
+                if (result) {
+                    logger.info("Add a new hbase mapping config: {} to canal adapter",
+                            file.getName());
+                }
             } catch (Exception e) {
                 logger.error(e.getMessage(), e);
             }
@@ -97,10 +94,7 @@ public class HbaseConfigMonitor {
                         return;
                     }
                     config.validate();
-                    if (hbaseAdapter.getHbaseMapping().containsKey(file.getName())) {
-                        deleteConfigFromCache(file);
-                    }
-                    addConfigToCache(file, config);
+                    hbaseAdapter.updateConfig(file.getName(), config);
                 }
             } catch (Exception e) {
                 logger.error(e.getMessage(), e);
@@ -113,33 +107,12 @@ public class HbaseConfigMonitor {
 
             try {
                 if (hbaseAdapter.getHbaseMapping().containsKey(file.getName())) {
-                    deleteConfigFromCache(file);
-
+                    hbaseAdapter.deleteConfig(file.getName());
                     logger.info("Delete a hbase mapping config: {} of canal adapter", file.getName());
                 }
             } catch (Exception e) {
                 logger.error(e.getMessage(), e);
             }
         }
-
-        private void addConfigToCache(File file, MappingConfig config) {
-            hbaseAdapter.getHbaseMapping().put(file.getName(), config);
-            Map<String, MappingConfig> configMap = hbaseAdapter.getMappingConfigCache()
-                .computeIfAbsent(StringUtils.trimToEmpty(config.getDestination()) + "_"
-                                 + config.getHbaseMapping().getDatabase() + "-" + config.getHbaseMapping().getTable(),
-                    k1 -> new HashMap<>());
-            configMap.put(file.getName(), config);
-        }
-
-        private void deleteConfigFromCache(File file) {
-
-            hbaseAdapter.getHbaseMapping().remove(file.getName());
-            for (Map<String, MappingConfig> configMap : hbaseAdapter.getMappingConfigCache().values()) {
-                if (configMap != null) {
-                    configMap.remove(file.getName());
-                }
-            }
-
-        }
     }
 }