Merge pull request #1132 from rewerma/master

rdb adapter整理

client-adapter/README.md

@@ -146,6 +146,11 @@ spring:
 canal.conf:
   canalServerHost: 127.0.0.1:11111
   flatMessage: true
+  srcDataSources:
+    defaultDS:
+      url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true
+      username: root
+      password: 121212
   canalInstances:
   - instance: example
     adapterGroups:
@@ -155,17 +160,8 @@ canal.conf:
           hbase.zookeeper.quorum: slave1
           hbase.zookeeper.property.clientPort: 2181
           zookeeper.znode.parent: /hbase
-
-adapter.conf:
-  datasourceConfigs:
-    defaultDS:
-      url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true
-      username: root
-      password: 121212
-  adapterConfigs:
-  - hbase/mytest_person.yml 
 ```
-其中指定了一个HBase表映射文件: mytest_person.yml
+adapter将会自动加载 conf/hbase 下的所有.yml结尾的配置文件
 ### 3.2 适配器表映射文件
 修改 conf/hbase/mytest_person.yml文件:
 ```
@@ -251,7 +247,7 @@ create 'MYTEST.PERSON', {NAME=>'CF'}
 ```
 #### 启动canal-adapter启动器
 ```
-java -jar canal-adapter-launcher.jar
+bin/startup.sh
 ```
 #### 验证
 修改mysql mytest.person表的数据, 将会自动同步到HBase的MYTEST.PERSON表下面, 并会打出DML的log

client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java

@@ -1,41 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.support;
-
-import java.util.HashMap;
-import java.util.LinkedHashSet;
-import java.util.Map;
-import java.util.Set;
-
-/**
- * 适配器配置集合, 用于配置加载, 线程不安全
- *
- * @author rewerma @ 2018-10-20
- * @version 1.0.0
- */
-public class AdapterConfigs {
-
-    /**
-     * 类型下对应所有配置名, 如:
-     * hbase
-     *  ┗━ mytest_person.yml
-     *  ┗━ mytest_role.yml
-     *  ┗━ mytest_department.yml
-     */
-    private static Map<String, Set<String>> configs = new HashMap<>();
-
-    public static void put(String key, String value) {
-        Set<String> values = configs.get(key);
-        if (values == null) {
-            values = new LinkedHashSet<>();
-        }
-        values.add(value);
-        configs.put(key, values);
-    }
-
-    public static Set<String> get(String key) {
-        return configs.get(key);
-    }
-
-    public static void clear() {
-        configs.clear();
-    }
-}

client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java

@@ -13,6 +13,7 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.regex.Pattern;
 
+import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -40,7 +41,9 @@ public class ExtensionLoader<T> {
 
     private static final ConcurrentMap<Class<?>, Object>             EXTENSION_INSTANCES        = new ConcurrentHashMap<>();
 
-    private static final ConcurrentMap<String, Object>               EXTENSION_KEY_INSTANCES    = new ConcurrentHashMap<>();
+    private static final ConcurrentMap<String, Object>               EXTENSION_KEY_INSTANCE     = new ConcurrentHashMap<>();
+
+    private static final ConcurrentMap<String, List<?>>              EXTENSION_KEY_INSTANCES    = new ConcurrentHashMap<>();
 
     private final Class<?>                                           type;
 
@@ -124,10 +127,11 @@ public class ExtensionLoader<T> {
         if ("true".equals(name)) {
             return getDefaultExtension();
         }
-        Holder<Object> holder = cachedInstances.get(name + "-" + key);
+        String extKey = name + "-" + StringUtils.trimToEmpty(key);
+        Holder<Object> holder = cachedInstances.get(extKey);
         if (holder == null) {
-            cachedInstances.putIfAbsent(name + "-" + key, new Holder<>());
+            cachedInstances.putIfAbsent(extKey, new Holder<>());
-            holder = cachedInstances.get(name + "-" + key);
+            holder = cachedInstances.get(extKey);
         }
         Object instance = holder.get();
         if (instance == null) {
@@ -182,10 +186,10 @@ public class ExtensionLoader<T> {
                                             + ")  could not be instantiated: class could not be found");
         }
         try {
-            T instance = (T) EXTENSION_KEY_INSTANCES.get(name + "-" + key);
+            T instance = (T) EXTENSION_KEY_INSTANCE.get(name + "-" + key);
             if (instance == null) {
-                EXTENSION_KEY_INSTANCES.putIfAbsent(name + "-" + key, clazz.newInstance());
+                EXTENSION_KEY_INSTANCE.putIfAbsent(name + "-" + key, clazz.newInstance());
-                instance = (T) EXTENSION_KEY_INSTANCES.get(name + "-" + key);
+                instance = (T) EXTENSION_KEY_INSTANCE.get(name + "-" + key);
             }
             return instance;
         } catch (Throwable t) {
@@ -195,24 +199,6 @@ public class ExtensionLoader<T> {
         }
     }
 
-    @SuppressWarnings("unchecked")
-    // public T newInstance(String name) {
-    // Class<?> clazz = getExtensionClasses().get(name);
-    // if (clazz == null) {
-    // throw new IllegalStateException("Extension instance(name: " + name + ",
-    // class: " + type
-    // + ") could not be instantiated: class could not be found");
-    // }
-    // try {
-    // return (T) clazz.newInstance();
-    // } catch (Throwable t) {
-    // throw new IllegalStateException("Extension instance(name: " + name + ",
-    // class: " + type
-    // + ") could not be instantiated: " + t.getMessage(),
-    // t);
-    // }
-    // }
-
     private Map<String, Class<?>> getExtensionClasses() {
         Map<String, Class<?>> classes = cachedClasses.get();
         if (classes == null) {

client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MappingConfigsLoader.java

@@ -0,0 +1,46 @@
+package com.alibaba.otter.canal.client.adapter.support;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+
+public class MappingConfigsLoader {
+
+    public static Map<String, String> loadConfigs(String name) {
+        Map<String, String> configContentMap = new HashMap<>();
+
+        // 先取本地文件，再取类路径
+        File configDir = new File("../conf/" + name);
+        if (!configDir.exists()) {
+            URL url = MappingConfigsLoader.class.getClassLoader().getResource("");
+            if (url != null) {
+                configDir = new File(url.getPath() + name + File.separator);
+            }
+        }
+
+        File[] files = configDir.listFiles();
+        if (files != null) {
+            for (File file : files) {
+                String fileName = file.getName();
+                if (!fileName.endsWith(".yml")) {
+                    continue;
+                }
+                try (InputStream in = new FileInputStream(file)) {
+                    byte[] bytes = new byte[in.available()];
+                    in.read(bytes);
+                    String configContent = new String(bytes, StandardCharsets.UTF_8);
+                    configContentMap.put(fileName, configContent);
+                } catch (IOException e) {
+                    throw new RuntimeException("Read " + name + "mapping config: " + fileName + " error. ", e);
+                }
+            }
+        }
+
+        return configContentMap;
+    }
+}

client-adapter/elasticsearch/pom.xml

@@ -71,6 +71,26 @@
                     </execution>
                 </executions>
             </plugin>
+            <plugin>
+                <artifactId>maven-antrun-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>run</goal>
+                        </goals>
+                        <configuration>
+                            <tasks>
+                                <copy todir="${project.basedir}/../launcher/target/classes/es" overwrite="true" >
+                                    <fileset dir="${project.basedir}/target/classes/es" erroronmissingdir="true">
+                                        <include name="*.yml"/>
+                                    </fileset>
+                                </copy>
+                            </tasks>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
         </plugins>
     </build>
 

client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java

@@ -1,9 +1,12 @@
 package com.alibaba.otter.canal.client.adapter.es;
 
 import java.net.InetAddress;
+import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import javax.sql.DataSource;
 
@@ -13,10 +16,13 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.transport.client.PreBuiltTransportClient;
 
+import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.otter.canal.client.adapter.OuterAdapter;
 import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
 import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping;
 import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfigLoader;
+import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem;
+import com.alibaba.otter.canal.client.adapter.es.config.SqlParser;
 import com.alibaba.otter.canal.client.adapter.es.service.ESEtlService;
 import com.alibaba.otter.canal.client.adapter.es.service.ESSyncService;
 import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate;
@@ -31,9 +37,12 @@ import com.alibaba.otter.canal.client.adapter.support.*;
 @SPI("es")
 public class ESAdapter implements OuterAdapter {
 
-    private TransportClient transportClient;
+    private Map<String, ESSyncConfig>       esSyncConfig        = new LinkedHashMap<>(); // 文件名对应配置
+    private Map<String, List<ESSyncConfig>> dbTableEsSyncConfig = new LinkedHashMap<>(); // schema-table对应配置
 
-    private ESSyncService   esSyncService;
+    private TransportClient                 transportClient;
+
+    private ESSyncService                   esSyncService;
 
     public TransportClient getTransportClient() {
         return transportClient;
@@ -43,10 +52,48 @@ public class ESAdapter implements OuterAdapter {
         return esSyncService;
     }
 
+    public Map<String, ESSyncConfig> getEsSyncConfig() {
+        return esSyncConfig;
+    }
+
+    public Map<String, List<ESSyncConfig>> getDbTableEsSyncConfig() {
+        return dbTableEsSyncConfig;
+    }
+
     @Override
     public void init(OuterAdapterConfig configuration) {
         try {
-            ESSyncConfigLoader.load();
+            Map<String, ESSyncConfig> esSyncConfigTmp = ESSyncConfigLoader.load();
+            // 过滤不匹配的key的配置
+            esSyncConfigTmp.forEach((key, config) -> {
+                if ((config.getOuterAdapterKey() == null && configuration.getKey() == null)
+                    || (config.getOuterAdapterKey() != null
+                        && config.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) {
+                    esSyncConfig.put(key, config);
+                }
+            });
+
+            for (ESSyncConfig config : esSyncConfig.values()) {
+                SchemaItem schemaItem = SqlParser.parse(config.getEsMapping().getSql());
+                config.getEsMapping().setSchemaItem(schemaItem);
+
+                DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
+                if (dataSource == null || dataSource.getUrl() == null) {
+                    throw new RuntimeException("No data source found: " + config.getDataSourceKey());
+                }
+                Pattern pattern = Pattern.compile(".*:(.*)://.*/(.*)\\?.*$");
+                Matcher matcher = pattern.matcher(dataSource.getUrl());
+                if (!matcher.find()) {
+                    throw new RuntimeException("Not found the schema of jdbc-url: " + config.getDataSourceKey());
+                }
+                String schema = matcher.group(2);
+
+                schemaItem.getAliasTableItems().values().forEach(tableItem -> {
+                    List<ESSyncConfig> esSyncConfigs = dbTableEsSyncConfig
+                        .computeIfAbsent(schema + "-" + tableItem.getTableName(), k -> new ArrayList<>());
+                    esSyncConfigs.add(config);
+                });
+            }
 
             Map<String, String> properties = configuration.getProperties();
             Settings.Builder settingBuilder = Settings.builder();
@@ -68,13 +115,16 @@ public class ESAdapter implements OuterAdapter {
 
     @Override
     public void sync(Dml dml) {
-        esSyncService.sync(dml);
+        String database = dml.getDatabase();
+        String table = dml.getTable();
+        List<ESSyncConfig> esSyncConfigs = dbTableEsSyncConfig.get(database + "-" + table);
+        esSyncService.sync(esSyncConfigs, dml);
     }
 
     @Override
     public EtlResult etl(String task, List<String> params) {
         EtlResult etlResult = new EtlResult();
-        ESSyncConfig config = ESSyncConfigLoader.getEsSyncConfig().get(task);
+        ESSyncConfig config = esSyncConfig.get(task);
         if (config != null) {
             DataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
             ESEtlService esEtlService = new ESEtlService(transportClient, config);
@@ -89,7 +139,7 @@ public class ESAdapter implements OuterAdapter {
             StringBuilder resultMsg = new StringBuilder();
             boolean resSuccess = true;
             // ds不为空说明传入的是datasourceKey
-            for (ESSyncConfig configTmp : ESSyncConfigLoader.getEsSyncConfig().values()) {
+            for (ESSyncConfig configTmp : esSyncConfig.values()) {
                 // 取所有的destination为task的配置
                 if (configTmp.getDestination().equals(task)) {
                     ESEtlService esEtlService = new ESEtlService(transportClient, configTmp);
@@ -119,7 +169,7 @@ public class ESAdapter implements OuterAdapter {
 
     @Override
     public Map<String, Object> count(String task) {
-        ESSyncConfig config = ESSyncConfigLoader.getEsSyncConfig().get(task);
+        ESSyncConfig config = esSyncConfig.get(task);
         ESMapping mapping = config.getEsMapping();
         SearchResponse response = transportClient.prepareSearch(mapping.get_index())
             .setTypes(mapping.get_type())
@@ -142,7 +192,7 @@ public class ESAdapter implements OuterAdapter {
 
     @Override
     public String getDestination(String task) {
-        ESSyncConfig config = ESSyncConfigLoader.getEsSyncConfig().get(task);
+        ESSyncConfig config = esSyncConfig.get(task);
         if (config != null) {
             return config.getDestination();
         }

client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java

@@ -13,9 +13,11 @@ import java.util.Map;
  */
 public class ESSyncConfig {
 
-    private String    dataSourceKey; // 数据源key
+    private String    dataSourceKey;   // 数据源key
 
-    private String    destination;   // canal destination
+    private String    outerAdapterKey; // adapter key
+
+    private String    destination;     // canal destination
 
     private ESMapping esMapping;
 
@@ -42,6 +44,14 @@ public class ESSyncConfig {
         this.dataSourceKey = dataSourceKey;
     }
 
+    public String getOuterAdapterKey() {
+        return outerAdapterKey;
+    }
+
+    public void setOuterAdapterKey(String outerAdapterKey) {
+        this.outerAdapterKey = outerAdapterKey;
+    }
+
     public String getDestination() {
         return destination;
     }
@@ -67,15 +77,15 @@ public class ESSyncConfig {
         private String              parent;
         private String              sql;
         // 对象字段, 例: objFields:
-        //              - _labels: array:;
+        // - _labels: array:;
-        private Map<String, String> objFields     = new LinkedHashMap<>();
+        private Map<String, String> objFields       = new LinkedHashMap<>();
         private List<String>        skips           = new ArrayList<>();
         private int                 commitBatch     = 1000;
         private String              etlCondition;
-        private boolean             syncByTimestamp = false;                 // 是否按时间戳定时同步
+        private boolean             syncByTimestamp = false;                // 是否按时间戳定时同步
-        private Long                syncInterval;                            // 同步时间间隔
+        private Long                syncInterval;                           // 同步时间间隔
 
-        private SchemaItem          schemaItem;                              // sql解析结果模型
+        private SchemaItem          schemaItem;                             // sql解析结果模型
 
         public String get_index() {
             return _index;

client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java

@@ -1,25 +1,13 @@
 package com.alibaba.otter.canal.client.adapter.es.config;
 
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.Collection;
 import java.util.LinkedHashMap;
-import java.util.List;
 import java.util.Map;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.yaml.snakeyaml.Yaml;
 
-import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
-import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
-import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 
 /**
  * ES 配置装载器
@@ -29,103 +17,25 @@ import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
  */
 public class ESSyncConfigLoader {
 
-    private static Logger                                   logger              = LoggerFactory
+    private static Logger logger = LoggerFactory.getLogger(ESSyncConfigLoader.class);
-        .getLogger(ESSyncConfigLoader.class);
 
-    private static final String                             BASE_PATH           = "es";
+    public static synchronized Map<String, ESSyncConfig> load() {
-
-    private static volatile Map<String, ESSyncConfig>       esSyncConfig        = new LinkedHashMap<>(); // 文件名对应配置
-    private static volatile Map<String, List<ESSyncConfig>> dbTableEsSyncConfig = new LinkedHashMap<>(); // schema-table对应配置
-
-    public static Map<String, ESSyncConfig> getEsSyncConfig() {
-        return esSyncConfig;
-    }
-
-    public static Map<String, List<ESSyncConfig>> getDbTableEsSyncConfig() {
-        return dbTableEsSyncConfig;
-    }
-
-    public static synchronized void load() {
         logger.info("## Start loading es mapping config ... ");
-        Collection<String> configs = AdapterConfigs.get("es");
-        if (configs == null) {
-            return;
-        }
-        for (String c : configs) {
-            if (c == null) {
-                continue;
-            }
-            c = c.trim();
-            if (c.equals("") || c.startsWith("#")) {
-                continue;
-            }
 
-            ESSyncConfig config;
+        Map<String, ESSyncConfig> esSyncConfig = new LinkedHashMap<>();
-            String configContent = null;
-
-            if (c.endsWith(".yml")) {
-                configContent = readConfigContent(BASE_PATH + "/" + c);
-            }
-
-            config = new Yaml().loadAs(configContent, ESSyncConfig.class);
 
+        Map<String, String> configContentMap = MappingConfigsLoader.loadConfigs("es");
+        configContentMap.forEach((fileName, content) -> {
+            ESSyncConfig config = new Yaml().loadAs(content, ESSyncConfig.class);
             try {
                 config.validate();
-                SchemaItem schemaItem = SqlParser.parse(config.getEsMapping().getSql());
-                config.getEsMapping().setSchemaItem(schemaItem);
-
-                DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
-                if (dataSource == null || dataSource.getUrl() == null) {
-                    throw new RuntimeException("No data source found: " + config.getDataSourceKey());
-                }
-                Pattern pattern = Pattern.compile(".*:(.*)://.*/(.*)\\?.*$");
-                Matcher matcher = pattern.matcher(dataSource.getUrl());
-                if (!matcher.find()) {
-                    throw new RuntimeException("Not found the schema of jdbc-url: " + config.getDataSourceKey());
-                }
-                String schema = matcher.group(2);
-
-                schemaItem.getAliasTableItems().values().forEach(tableItem -> {
-                    List<ESSyncConfig> esSyncConfigs = dbTableEsSyncConfig
-                        .computeIfAbsent(schema + "-" + tableItem.getTableName(), k -> new ArrayList<>());
-                    esSyncConfigs.add(config);
-                });
             } catch (Exception e) {
-                throw new RuntimeException("ERROR Config: " + c, e);
+                throw new RuntimeException("ERROR Config: " + fileName + " " + e.getMessage(), e);
             }
-            esSyncConfig.put(c, config);
+            esSyncConfig.put(fileName, config);
-        }
+        });
 
         logger.info("## ES mapping config loaded");
-    }
+        return esSyncConfig;
-
-    private static String readConfigContent(String config) {
-        InputStream in = null;
-        try {
-            // 先取本地文件，再取类路径
-            File configFile = new File("../conf/" + config);
-            if (configFile.exists()) {
-                in = new FileInputStream(configFile);
-            } else {
-                    in = ESSyncConfigLoader.class.getClassLoader().getResourceAsStream(config);
-            }
-            if (in == null) {
-                throw new RuntimeException("Config file: " + config + " not found.");
-            }
-
-            byte[] bytes = new byte[in.available()];
-            in.read(bytes);
-            return new String(bytes, StandardCharsets.UTF_8);
-        } catch (IOException e) {
-            throw new RuntimeException("Read es mapping config error ", e);
-        } finally {
-            try {
-                if (in != null) {
-                    in.close();
-                }
-            } catch (IOException e) {
-                // ignore
-            }
-        }
     }
 }

client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java

@@ -39,11 +39,8 @@ public class ESSyncService {
         this.esTemplate = esTemplate;
     }
 
-    public void sync(Dml dml) {
+    public void sync(List<ESSyncConfig> esSyncConfigs, Dml dml) {
         long begin = System.currentTimeMillis();
-        String database = dml.getDatabase();
-        String table = dml.getTable();
-        List<ESSyncConfig> esSyncConfigs = ESSyncConfigLoader.getDbTableEsSyncConfig().get(database + "-" + table);
         if (esSyncConfigs != null) {
             if (logger.isTraceEnabled()) {
                 logger.trace("Destination: {}, database:{}, table:{}, type:{}, effect index count: {}",

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ConfigLoadTest.java

@@ -1,6 +1,5 @@
 package com.alibaba.otter.canal.client.adapter.es.test;
 
-import java.util.List;
 import java.util.Map;
 
 import org.junit.Assert;
@@ -9,22 +8,20 @@ import org.junit.Test;
 
 import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
 import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfigLoader;
-import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
 import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 
 public class ConfigLoadTest {
 
     @Before
     public void before() {
-        AdapterConfigs.put("es", "mytest_user.yml");
+        // AdapterConfigs.put("es", "mytest_user.yml");
         // 加载数据源连接池
         DatasourceConfig.DATA_SOURCES.put("defaultDS", TestConstant.dataSource);
     }
 
     @Test
     public void testLoad() {
-        ESSyncConfigLoader.load();
+        Map<String, ESSyncConfig> configMap = ESSyncConfigLoader.load();
-        Map<String, ESSyncConfig> configMap = ESSyncConfigLoader.getEsSyncConfig();
         ESSyncConfig config = configMap.get("mytest_user.yml");
         Assert.assertNotNull(config);
         Assert.assertEquals("defaultDS", config.getDataSourceKey());
@@ -34,7 +31,8 @@ public class ConfigLoadTest {
         Assert.assertEquals("id", esMapping.get_id());
         Assert.assertNotNull(esMapping.getSql());
 
-        Map<String, List<ESSyncConfig>> dbTableEsSyncConfig = ESSyncConfigLoader.getDbTableEsSyncConfig();
+        // Map<String, List<ESSyncConfig>> dbTableEsSyncConfig =
-        Assert.assertFalse(dbTableEsSyncConfig.isEmpty());
+        // ESSyncConfigLoader.getDbTableEsSyncConfig();
+        // Assert.assertFalse(dbTableEsSyncConfig.isEmpty());
     }
 }

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java

@@ -2,27 +2,25 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync;
 
 import java.util.*;
 
+import javax.sql.DataSource;
+
 import org.elasticsearch.action.get.GetResponse;
-import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
-import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.otter.canal.client.adapter.es.ESAdapter;
-import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
+import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
 import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 import com.alibaba.otter.canal.client.adapter.support.Dml;
 
-import javax.sql.DataSource;
-
 public class LabelSyncJoinSub2Test {
 
     private ESAdapter esAdapter;
 
     @Before
     public void init() {
-        AdapterConfigs.put("es", "mytest_user_join_sub2.yml");
+        // AdapterConfigs.put("es", "mytest_user_join_sub2.yml");
         esAdapter = Common.init();
     }
 
@@ -32,9 +30,9 @@ public class LabelSyncJoinSub2Test {
     @Test
     public void test01() {
         DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
-        Common.sqlExe(ds,"delete from label where id=1 or id=2");
+        Common.sqlExe(ds, "delete from label where id=1 or id=2");
-        Common.sqlExe(ds,"insert into label (id,user_id,label) values (1,1,'a')");
+        Common.sqlExe(ds, "insert into label (id,user_id,label) values (1,1,'a')");
-        Common.sqlExe(ds,"insert into label (id,user_id,label) values (2,1,'b')");
+        Common.sqlExe(ds, "insert into label (id,user_id,label) values (2,1,'b')");
 
         Dml dml = new Dml();
         dml.setDestination("example");
@@ -46,12 +44,15 @@ public class LabelSyncJoinSub2Test {
         Map<String, Object> data = new LinkedHashMap<>();
         dataList.add(data);
         data.put("id", 2L);
-        data.put("user_id",1L);
+        data.put("user_id", 1L);
         data.put("label", "b");
-
         dml.setData(dataList);
 
-        esAdapter.getEsSyncService().sync(dml);
+        String database = dml.getDatabase();
+        String table = dml.getTable();
+        List<ESSyncConfig> esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("b;a_", response.getSource().get("_labels"));
@@ -63,7 +64,7 @@ public class LabelSyncJoinSub2Test {
     @Test
     public void test02() {
         DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
-        Common.sqlExe(ds,"update label set label='aa' where id=1");
+        Common.sqlExe(ds, "update label set label='aa' where id=1");
 
         Dml dml = new Dml();
         dml.setDestination("example");
@@ -75,7 +76,7 @@ public class LabelSyncJoinSub2Test {
         Map<String, Object> data = new LinkedHashMap<>();
         dataList.add(data);
         data.put("id", 1L);
-        data.put("user_id",1L);
+        data.put("user_id", 1L);
         data.put("label", "aa");
         dml.setData(dataList);
 
@@ -85,7 +86,11 @@ public class LabelSyncJoinSub2Test {
         old.put("label", "v");
         dml.setOld(oldList);
 
-        esAdapter.getEsSyncService().sync(dml);
+        String database = dml.getDatabase();
+        String table = dml.getTable();
+        List<ESSyncConfig> esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("b;aa_", response.getSource().get("_labels"));
@@ -97,7 +102,7 @@ public class LabelSyncJoinSub2Test {
     @Test
     public void test03() {
         DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
-        Common.sqlExe(ds,"delete from label where id=1");
+        Common.sqlExe(ds, "delete from label where id=1");
 
         Dml dml = new Dml();
         dml.setDestination("example");
@@ -109,12 +114,15 @@ public class LabelSyncJoinSub2Test {
         Map<String, Object> data = new LinkedHashMap<>();
         dataList.add(data);
         data.put("id", 1L);
-        data.put("user_id",1L);
+        data.put("user_id", 1L);
         data.put("label", "a");
-
         dml.setData(dataList);
 
-        esAdapter.getEsSyncService().sync(dml);
+        String database = dml.getDatabase();
+        String table = dml.getTable();
+        List<ESSyncConfig> esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("b_", response.getSource().get("_labels"));

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java

@@ -2,27 +2,25 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync;
 
 import java.util.*;
 
+import javax.sql.DataSource;
+
 import org.elasticsearch.action.get.GetResponse;
-import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
-import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.otter.canal.client.adapter.es.ESAdapter;
-import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
+import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
 import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 import com.alibaba.otter.canal.client.adapter.support.Dml;
 
-import javax.sql.DataSource;
-
 public class LabelSyncJoinSubTest {
 
     private ESAdapter esAdapter;
 
     @Before
     public void init() {
-        AdapterConfigs.put("es", "mytest_user_join_sub.yml");
+        // AdapterConfigs.put("es", "mytest_user_join_sub.yml");
         esAdapter = Common.init();
     }
 
@@ -32,9 +30,9 @@ public class LabelSyncJoinSubTest {
     @Test
     public void test01() {
         DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
-        Common.sqlExe(ds,"delete from label where id=1 or id=2");
+        Common.sqlExe(ds, "delete from label where id=1 or id=2");
-        Common.sqlExe(ds,"insert into label (id,user_id,label) values (1,1,'a')");
+        Common.sqlExe(ds, "insert into label (id,user_id,label) values (1,1,'a')");
-        Common.sqlExe(ds,"insert into label (id,user_id,label) values (2,1,'b')");
+        Common.sqlExe(ds, "insert into label (id,user_id,label) values (2,1,'b')");
 
         Dml dml = new Dml();
         dml.setDestination("example");
@@ -46,12 +44,15 @@ public class LabelSyncJoinSubTest {
         Map<String, Object> data = new LinkedHashMap<>();
         dataList.add(data);
         data.put("id", 2L);
-        data.put("user_id",1L);
+        data.put("user_id", 1L);
         data.put("label", "b");
-
         dml.setData(dataList);
 
-        esAdapter.getEsSyncService().sync(dml);
+        String database = dml.getDatabase();
+        String table = dml.getTable();
+        List<ESSyncConfig> esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("b;a", response.getSource().get("_labels"));
@@ -63,7 +64,7 @@ public class LabelSyncJoinSubTest {
     @Test
     public void test02() {
         DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
-        Common.sqlExe(ds,"update label set label='aa' where id=1");
+        Common.sqlExe(ds, "update label set label='aa' where id=1");
 
         Dml dml = new Dml();
         dml.setDestination("example");
@@ -75,7 +76,7 @@ public class LabelSyncJoinSubTest {
         Map<String, Object> data = new LinkedHashMap<>();
         dataList.add(data);
         data.put("id", 1L);
-        data.put("user_id",1L);
+        data.put("user_id", 1L);
         data.put("label", "aa");
         dml.setData(dataList);
 
@@ -85,7 +86,11 @@ public class LabelSyncJoinSubTest {
         old.put("label", "a");
         dml.setOld(oldList);
 
-        esAdapter.getEsSyncService().sync(dml);
+        String database = dml.getDatabase();
+        String table = dml.getTable();
+        List<ESSyncConfig> esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("b;aa", response.getSource().get("_labels"));
@@ -97,7 +102,7 @@ public class LabelSyncJoinSubTest {
     @Test
     public void test03() {
         DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
-        Common.sqlExe(ds,"delete from label where id=1");
+        Common.sqlExe(ds, "delete from label where id=1");
 
         Dml dml = new Dml();
         dml.setDestination("example");
@@ -109,12 +114,15 @@ public class LabelSyncJoinSubTest {
         Map<String, Object> data = new LinkedHashMap<>();
         dataList.add(data);
         data.put("id", 1L);
-        data.put("user_id",1L);
+        data.put("user_id", 1L);
         data.put("label", "a");
-
         dml.setData(dataList);
 
-        esAdapter.getEsSyncService().sync(dml);
+        String database = dml.getDatabase();
+        String table = dml.getTable();
+        List<ESSyncConfig> esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("b", response.getSource().get("_labels"));

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java

@@ -2,27 +2,25 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync;
 
 import java.util.*;
 
+import javax.sql.DataSource;
+
 import org.elasticsearch.action.get.GetResponse;
-import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
-import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.otter.canal.client.adapter.es.ESAdapter;
-import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
+import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
 import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 import com.alibaba.otter.canal.client.adapter.support.Dml;
 
-import javax.sql.DataSource;
-
 public class RoleSyncJoinOne2Test {
 
     private ESAdapter esAdapter;
 
     @Before
     public void init() {
-        AdapterConfigs.put("es", "mytest_user_join_one2.yml");
+        // AdapterConfigs.put("es", "mytest_user_join_one2.yml");
         esAdapter = Common.init();
     }
 
@@ -32,8 +30,8 @@ public class RoleSyncJoinOne2Test {
     @Test
     public void test01() {
         DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
-        Common.sqlExe(ds,"delete from role where id=1");
+        Common.sqlExe(ds, "delete from role where id=1");
-        Common.sqlExe(ds,"insert into role (id,role_name) values (1,'admin')");
+        Common.sqlExe(ds, "insert into role (id,role_name) values (1,'admin')");
 
         Dml dml = new Dml();
         dml.setDestination("example");
@@ -46,10 +44,13 @@ public class RoleSyncJoinOne2Test {
         dataList.add(data);
         data.put("id", 1L);
         data.put("role_name", "admin");
-
         dml.setData(dataList);
 
-        esAdapter.getEsSyncService().sync(dml);
+        String database = dml.getDatabase();
+        String table = dml.getTable();
+        List<ESSyncConfig> esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("admin_", response.getSource().get("_role_name"));
@@ -61,7 +62,7 @@ public class RoleSyncJoinOne2Test {
     @Test
     public void test02() {
         DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
-        Common.sqlExe(ds,"update role set role_name='admin3' where id=1");
+        Common.sqlExe(ds, "update role set role_name='admin3' where id=1");
 
         Dml dml = new Dml();
         dml.setDestination("example");
@@ -82,7 +83,11 @@ public class RoleSyncJoinOne2Test {
         old.put("role_name", "admin");
         dml.setOld(oldList);
 
-        esAdapter.getEsSyncService().sync(dml);
+        String database = dml.getDatabase();
+        String table = dml.getTable();
+        List<ESSyncConfig> esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("admin3_", response.getSource().get("_role_name"));

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java

@@ -10,7 +10,7 @@ import org.junit.Before;
 import org.junit.Test;
 
 import com.alibaba.otter.canal.client.adapter.es.ESAdapter;
-import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
+import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
 import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 import com.alibaba.otter.canal.client.adapter.support.Dml;
 
@@ -20,7 +20,7 @@ public class RoleSyncJoinOneTest {
 
     @Before
     public void init() {
-        AdapterConfigs.put("es", "mytest_user_join_one.yml");
+        // AdapterConfigs.put("es", "mytest_user_join_one.yml");
         esAdapter = Common.init();
     }
 
@@ -47,7 +47,11 @@ public class RoleSyncJoinOneTest {
 
         dml.setData(dataList);
 
-        esAdapter.getEsSyncService().sync(dml);
+        String database = dml.getDatabase();
+        String table = dml.getTable();
+        List<ESSyncConfig> esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("admin", response.getSource().get("_role_name"));
@@ -80,7 +84,11 @@ public class RoleSyncJoinOneTest {
         old.put("role_name", "admin");
         dml.setOld(oldList);
 
-        esAdapter.getEsSyncService().sync(dml);
+        String database = dml.getDatabase();
+        String table = dml.getTable();
+        List<ESSyncConfig> esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("admin2", response.getSource().get("_role_name"));
@@ -96,28 +104,32 @@ public class RoleSyncJoinOneTest {
         Common.sqlExe(ds, "insert into role (id,role_name) values (2,'operator')");
         Common.sqlExe(ds, "update user set role_id=2 where id=1");
 
-         Dml dml = new Dml();
+        Dml dml = new Dml();
-         dml.setDestination("example");
+        dml.setDestination("example");
-         dml.setTs(new Date().getTime());
+        dml.setTs(new Date().getTime());
-         dml.setType("UPDATE");
+        dml.setType("UPDATE");
-         dml.setDatabase("mytest");
+        dml.setDatabase("mytest");
-         dml.setTable("user");
+        dml.setTable("user");
-         List<Map<String, Object>> dataList = new ArrayList<>();
+        List<Map<String, Object>> dataList = new ArrayList<>();
-         Map<String, Object> data = new LinkedHashMap<>();
+        Map<String, Object> data = new LinkedHashMap<>();
-         dataList.add(data);
+        dataList.add(data);
-         data.put("id", 1L);
+        data.put("id", 1L);
-         data.put("role_id", 2L);
+        data.put("role_id", 2L);
-         dml.setData(dataList);
+        dml.setData(dataList);
-         List<Map<String, Object>> oldList = new ArrayList<>();
+        List<Map<String, Object>> oldList = new ArrayList<>();
-         Map<String, Object> old = new LinkedHashMap<>();
+        Map<String, Object> old = new LinkedHashMap<>();
-         oldList.add(old);
+        oldList.add(old);
-         old.put("role_id", 1L);
+        old.put("role_id", 1L);
-         dml.setOld(oldList);
+        dml.setOld(oldList);
-         esAdapter.getEsSyncService().sync(dml);
 
-         GetResponse response =
+        String database = dml.getDatabase();
-         esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
+        String table = dml.getTable();
-         Assert.assertEquals("operator", response.getSource().get("_role_name"));
+        List<ESSyncConfig> esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml);
+
+        GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
+        Assert.assertEquals("operator", response.getSource().get("_role_name"));
 
         Common.sqlExe(ds, "update user set role_id=1 where id=1");
 
@@ -138,7 +150,8 @@ public class RoleSyncJoinOneTest {
         oldList2.add(old2);
         old2.put("role_id", 2L);
         dml2.setOld(oldList2);
-        esAdapter.getEsSyncService().sync(dml2);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml2);
 
         GetResponse response2 = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("admin2", response2.getSource().get("_role_name"));
@@ -166,7 +179,11 @@ public class RoleSyncJoinOneTest {
 
         dml.setData(dataList);
 
-        esAdapter.getEsSyncService().sync(dml);
+        String database = dml.getDatabase();
+        String table = dml.getTable();
+        List<ESSyncConfig> esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertNull(response.getSource().get("_role_name"));

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java

@@ -5,14 +5,12 @@ import java.util.*;
 import javax.sql.DataSource;
 
 import org.elasticsearch.action.get.GetResponse;
-import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
-import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.otter.canal.client.adapter.es.ESAdapter;
-import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
+import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
 import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 import com.alibaba.otter.canal.client.adapter.support.Dml;
 
@@ -22,7 +20,7 @@ public class UserSyncJoinOneTest {
 
     @Before
     public void init() {
-        AdapterConfigs.put("es", "mytest_user_join_one.yml");
+//        AdapterConfigs.put("es", "mytest_user_join_one.yml");
         esAdapter = Common.init();
     }
 
@@ -32,8 +30,8 @@ public class UserSyncJoinOneTest {
     @Test
     public void test01() {
         DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
-        Common.sqlExe(ds,"delete from user where id=1");
+        Common.sqlExe(ds, "delete from user where id=1");
-        Common.sqlExe(ds,"insert into user (id,name,role_id) values (1,'Eric',1)");
+        Common.sqlExe(ds, "insert into user (id,name,role_id) values (1,'Eric',1)");
 
         Dml dml = new Dml();
         dml.setDestination("example");
@@ -48,10 +46,13 @@ public class UserSyncJoinOneTest {
         data.put("name", "Eric");
         data.put("role_id", 1L);
         data.put("c_time", new Date());
-
         dml.setData(dataList);
 
-        esAdapter.getEsSyncService().sync(dml);
+        String database = dml.getDatabase();
+        String table = dml.getTable();
+        List<ESSyncConfig> esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("Eric_", response.getSource().get("_name"));
@@ -63,7 +64,7 @@ public class UserSyncJoinOneTest {
     @Test
     public void test02() {
         DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
-        Common.sqlExe(ds,"update user set name='Eric2' where id=1");
+        Common.sqlExe(ds, "update user set name='Eric2' where id=1");
 
         Dml dml = new Dml();
         dml.setDestination("example");
@@ -83,7 +84,11 @@ public class UserSyncJoinOneTest {
         old.put("name", "Eric");
         dml.setOld(oldList);
 
-        esAdapter.getEsSyncService().sync(dml);
+        String database = dml.getDatabase();
+        String table = dml.getTable();
+        List<ESSyncConfig> esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("Eric2_", response.getSource().get("_name"));

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java

@@ -3,15 +3,12 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync;
 import java.util.*;
 
 import org.elasticsearch.action.get.GetResponse;
-import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
-import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.otter.canal.client.adapter.es.ESAdapter;
-import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
+import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
-import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 import com.alibaba.otter.canal.client.adapter.support.Dml;
 
 public class UserSyncSingleTest {
@@ -20,7 +17,7 @@ public class UserSyncSingleTest {
 
     @Before
     public void init() {
-        AdapterConfigs.put("es", "mytest_user_single.yml");
+        // AdapterConfigs.put("es", "mytest_user_single.yml");
         esAdapter = Common.init();
     }
 
@@ -42,10 +39,13 @@ public class UserSyncSingleTest {
         data.put("name", "Eric");
         data.put("role_id", 1L);
         data.put("c_time", new Date());
-
         dml.setData(dataList);
 
-        esAdapter.getEsSyncService().sync(dml);
+        String database = dml.getDatabase();
+        String table = dml.getTable();
+        List<ESSyncConfig> esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("Eric", response.getSource().get("_name"));
@@ -74,7 +74,11 @@ public class UserSyncSingleTest {
         old.put("name", "Eric");
         dml.setOld(oldList);
 
-        esAdapter.getEsSyncService().sync(dml);
+        String database = dml.getDatabase();
+        String table = dml.getTable();
+        List<ESSyncConfig> esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("Eric2", response.getSource().get("_name"));
@@ -98,10 +102,13 @@ public class UserSyncSingleTest {
         data.put("name", "Eric");
         data.put("role_id", 1L);
         data.put("c_time", new Date());
-
         dml.setData(dataList);
 
-        esAdapter.getEsSyncService().sync(dml);
+        String database = dml.getDatabase();
+        String table = dml.getTable();
+        List<ESSyncConfig> esSyncConfigs = esAdapter.getDbTableEsSyncConfig().get(database + "-" + table);
+
+        esAdapter.getEsSyncService().sync(esSyncConfigs, dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertNull(response.getSource());

client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one.yml

@@ -1,10 +0,0 @@
-dataSourceKey: defaultDS
-destination: example
-esMapping:
-  _index: mytest_user
-  _type: _doc
-  _id: _id
-  sql: "select a.id as _id, concat(a.name,'_') as _name, a.role_id as _role_id,
-        b.role_name as _role_name, a.c_time as _c_time from user a
-        left join role b on b.id=a.role_id"
-  commitBatch: 3000

client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one2.yml

@@ -1,10 +0,0 @@
-dataSourceKey: defaultDS
-destination: example
-esMapping:
-  _index: mytest_user
-  _type: _doc
-  _id: _id
-  sql: "select a.id as _id, concat(a.name,'_') as _name, a.role_id as _role_id,
-        concat(b.role_name,'_') as _role_name, a.c_time as _c_time from user a
-        left join role b on b.id=a.role_id"
-  commitBatch: 3000

client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub.yml

@@ -1,11 +0,0 @@
-dataSourceKey: defaultDS
-destination: example
-esMapping:
-  _index: mytest_user
-  _type: _doc
-  _id: _id
-  sql: "select a.id as _id, concat(a.name,'_') as _name, a.role_id as _role_id,
-        b.labels _labels, a.c_time as _c_time from user a
-        left join (select user_id, group_concat(label order by id desc separator ';') as labels from label
-        group by user_id) b on b.user_id=a.id"
-  commitBatch: 3000

client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub2.yml

@@ -1,11 +0,0 @@
-dataSourceKey: defaultDS
-destination: example
-esMapping:
-  _index: mytest_user
-  _type: _doc
-  _id: _id
-  sql: "select a.id as _id, concat(a.name,'_') as _name, a.role_id as _role_id,
-        concat(b.labels, '_') as _labels, a.c_time as _c_time from user a
-        left join (select user_id, group_concat(label order by id desc separator ';') as labels from label
-        group by user_id) b on b.user_id=a.id"
-  commitBatch: 3000

client-adapter/hbase/pom.xml

@@ -62,6 +62,26 @@
                     </execution>
                 </executions>
             </plugin>
+            <plugin>
+                <artifactId>maven-antrun-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>run</goal>
+                        </goals>
+                        <configuration>
+                            <tasks>
+                                <copy todir="${project.basedir}/../launcher/target/classes/hbase" overwrite="true" >
+                                    <fileset dir="${project.basedir}/target/classes/hbase" erroronmissingdir="true">
+                                        <include name="*.yml"/>
+                                    </fileset>
+                                </copy>
+                            </tasks>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
         </plugins>
     </build>
 

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

@@ -35,31 +35,32 @@ import com.alibaba.otter.canal.client.adapter.support.*;
 @SPI("hbase")
 public class HbaseAdapter implements OuterAdapter {
 
-    private static Logger                              logger             = LoggerFactory.getLogger(HbaseAdapter.class);
+    private static Logger              logger             = LoggerFactory.getLogger(HbaseAdapter.class);
 
-    private static volatile Map<String, MappingConfig> hbaseMapping       = null;                                       // 文件名对应配置
+    private Map<String, MappingConfig> hbaseMapping       = new HashMap<>();                            // 文件名对应配置
-    private static volatile Map<String, MappingConfig> mappingConfigCache = null;                                       // 库名-表名对应配置
+    private Map<String, MappingConfig> mappingConfigCache = new HashMap<>();                            // 库名-表名对应配置
 
-    private Connection                                 conn;
+    private Connection                 conn;
-    private HbaseSyncService                           hbaseSyncService;
+    private HbaseSyncService           hbaseSyncService;
-    private HbaseTemplate                              hbaseTemplate;
+    private HbaseTemplate              hbaseTemplate;
 
     @Override
     public void init(OuterAdapterConfig configuration) {
         try {
-            if (mappingConfigCache == null) {
+            Map<String, MappingConfig> hbaseMappingTmp = MappingConfigLoader.load();
-                synchronized (MappingConfig.class) {
+            // 过滤不匹配的key的配置
-                    if (mappingConfigCache == null) {
+            hbaseMappingTmp.forEach((key, mappingConfig) -> {
-                        hbaseMapping = MappingConfigLoader.load();
+                if ((mappingConfig.getOuterAdapterKey() == null && configuration.getKey() == null)
-                        mappingConfigCache = new HashMap<>();
+                    || (mappingConfig.getOuterAdapterKey() != null
-                        for (MappingConfig mappingConfig : hbaseMapping.values()) {
+                        && mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) {
-                            mappingConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "."
+                    hbaseMapping.put(key, mappingConfig);
-                                                   + mappingConfig.getHbaseMapping().getDatabase() + "."
-                                                   + mappingConfig.getHbaseMapping().getTable(),
-                                mappingConfig);
-                        }
-                    }
                 }
+            });
+            for (MappingConfig mappingConfig : hbaseMapping.values()) {
+                mappingConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "."
+                                       + mappingConfig.getHbaseMapping().getDatabase() + "."
+                                       + mappingConfig.getHbaseMapping().getTable(),
+                    mappingConfig);
             }
 
             Map<String, String> properties = configuration.getProperties();

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

@@ -10,11 +10,13 @@ import java.util.*;
  */
 public class MappingConfig {
 
-    private String       dataSourceKey; // 数据源key
+    private String       dataSourceKey;   // 数据源key
 
-    private String       destination;   // canal实例或MQ的topic
+    private String       outerAdapterKey; // adapter key
 
-    private HbaseMapping hbaseMapping;  // hbase映射配置
+    private String       destination;     // canal实例或MQ的topic
+
+    private HbaseMapping hbaseMapping;    // hbase映射配置
 
     public String getDataSourceKey() {
         return dataSourceKey;
@@ -24,6 +26,14 @@ public class MappingConfig {
         this.dataSourceKey = dataSourceKey;
     }
 
+    public String getOuterAdapterKey() {
+        return outerAdapterKey;
+    }
+
+    public void setOuterAdapterKey(String outerAdapterKey) {
+        this.outerAdapterKey = outerAdapterKey;
+    }
+
     public String getDestination() {
         return destination;
     }

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

@@ -1,20 +1,13 @@
 package com.alibaba.otter.canal.client.adapter.hbase.config;
 
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.charset.StandardCharsets;
-import java.util.Collection;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
-import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.yaml.snakeyaml.Yaml;
 
-import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
+import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
 
 /**
  * HBase表映射配置加载器
@@ -24,9 +17,7 @@ import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
  */
 public class MappingConfigLoader {
 
-    private static Logger       logger    = LoggerFactory.getLogger(MappingConfigLoader.class);
+    private static Logger logger = LoggerFactory.getLogger(MappingConfigLoader.class);
-
-    private static final String BASE_PATH = "hbase";
 
     /**
      * 加载HBase表映射配置
@@ -38,116 +29,18 @@ public class MappingConfigLoader {
 
         Map<String, MappingConfig> result = new LinkedHashMap<>();
 
-        Collection<String> configs = AdapterConfigs.get("hbase");
+        Map<String, String> configContentMap = MappingConfigsLoader.loadConfigs("hbase");
-        if (configs == null) {
+        configContentMap.forEach((fileName, content) -> {
-            return result;
+            MappingConfig config = new Yaml().loadAs(content, MappingConfig.class);
-        }
-        for (String c : configs) {
-            if (c == null) {
-                continue;
-            }
-            c = c.trim();
-            if (c.equals("") || c.startsWith("#")) {
-                continue;
-            }
-
-            MappingConfig config;
-            String configContent = null;
-
-            if (c.endsWith(".yml")) {
-                configContent = readConfigContent(BASE_PATH + "/" + c);
-            }
-
-            // 简单配置database.table@datasourcekey?rowKey=key1,key2
-            if (StringUtils.isEmpty(configContent)) {
-                String[] mapping = c.split("\\?");
-                String params = mapping.length == 2 ? mapping[1] : null;
-                String rowKey = null;
-                String srcMeta = mapping[0];
-                //
-                if (params != null) {
-                    for (String entry : params.split("&")) {
-                        if ("rowKey".equals(entry.split("=")[0])) {
-                            rowKey = entry.split("=")[1];
-                        }
-                    }
-                }
-                String dsKey = srcMeta.split("@").length == 2 ? srcMeta.split("@")[1] : null;
-                String[] dbTable;
-                if (dsKey == null) {
-                    dbTable = srcMeta.split("\\.");
-
-                } else {
-                    dbTable = srcMeta.split("@")[0].split("\\.");
-                }
-
-                if (dbTable.length == 2) {
-                    config = new MappingConfig();
-
-                    MappingConfig.HbaseMapping hbaseMapping = new MappingConfig.HbaseMapping();
-                    hbaseMapping.setHbaseTable(dbTable[0].toUpperCase() + "." + dbTable[1].toUpperCase());
-                    hbaseMapping.setAutoCreateTable(true);
-                    hbaseMapping.setDatabase(dbTable[0]);
-                    hbaseMapping.setTable(dbTable[1]);
-                    hbaseMapping.setMode(MappingConfig.Mode.PHOENIX);
-                    hbaseMapping.setRowKey(rowKey);
-                    // 有定义rowKey
-                    if (rowKey != null) {
-                        MappingConfig.ColumnItem columnItem = new MappingConfig.ColumnItem();
-                        columnItem.setRowKey(true);
-                        columnItem.setColumn(rowKey);
-                        hbaseMapping.setRowKeyColumn(columnItem);
-                    }
-                    config.setHbaseMapping(hbaseMapping);
-                    config.setDataSourceKey(dsKey);
-
-                } else {
-                    throw new RuntimeException(String.format("配置项[%s]内容为空, 或格式不符合database.table", c));
-                }
-
-            } else { // 配置文件配置
-                config = new Yaml().loadAs(configContent, MappingConfig.class);
-            }
-
             try {
                 config.validate();
             } catch (Exception e) {
-                throw new RuntimeException("ERROR Config: " + c + " " + e.getMessage(), e);
+                throw new RuntimeException("ERROR load Config: " + fileName + " " + e.getMessage(), e);
             }
-            result.put(c, config);
+            result.put(fileName, config);
-        }
+        });
 
         logger.info("## Hbase mapping config loaded");
         return result;
     }
-
-    public static String readConfigContent(String config) {
-        InputStream in = null;
-        try {
-            // 先取本地文件，再取类路径
-            File configFile = new File("../conf/" + config);
-            if (configFile.exists()) {
-                in = new FileInputStream(configFile);
-            } else {
-                in = MappingConfigLoader.class.getClassLoader().getResourceAsStream(config);
-            }
-            if (in == null) {
-                throw new RuntimeException("Config file not found.");
-            }
-
-            byte[] bytes = new byte[in.available()];
-            in.read(bytes);
-            return new String(bytes, StandardCharsets.UTF_8);
-        } catch (IOException e) {
-            throw new RuntimeException("Read hbase mapping config error. ", e);
-        } finally {
-            try {
-                if (in != null) {
-                    in.close();
-                }
-            } catch (IOException e) {
-                // ignore
-            }
-        }
-    }
 }

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java

@@ -1,9 +1,13 @@
 package com.alibaba.otter.canal.adapter.launcher.config;
 
+import java.sql.SQLException;
 import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
+import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.stereotype.Component;
 
@@ -19,7 +23,9 @@ import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
 @ConfigurationProperties(prefix = "canal.conf")
 public class AdapterCanalConfig extends CanalClientConfig {
 
-    public final Set<String> DESTINATIONS = new LinkedHashSet<>();
+    public final Set<String>              DESTINATIONS = new LinkedHashSet<>();
+
+    private Map<String, DatasourceConfig> srcDataSources;
 
     @Override
     public void setCanalInstances(List<CanalInstance> canalInstances) {
@@ -48,4 +54,37 @@ public class AdapterCanalConfig extends CanalClientConfig {
             }
         }
     }
+
+    public Map<String, DatasourceConfig> getSrcDataSources() {
+        return srcDataSources;
+    }
+
+    public void setSrcDataSources(Map<String, DatasourceConfig> srcDataSources) {
+        this.srcDataSources = srcDataSources;
+
+        if (srcDataSources != null) {
+            for (Map.Entry<String, DatasourceConfig> entry : srcDataSources.entrySet()) {
+                DatasourceConfig datasourceConfig = entry.getValue();
+                // 加载数据源连接池
+                DruidDataSource ds = new DruidDataSource();
+                ds.setDriverClassName(datasourceConfig.getDriver());
+                ds.setUrl(datasourceConfig.getUrl());
+                ds.setUsername(datasourceConfig.getUsername());
+                ds.setPassword(datasourceConfig.getPassword());
+                ds.setInitialSize(1);
+                ds.setMinIdle(1);
+                ds.setMaxActive(datasourceConfig.getMaxActive());
+                ds.setMaxWait(60000);
+                ds.setTimeBetweenEvictionRunsMillis(60000);
+                ds.setMinEvictableIdleTimeMillis(300000);
+                ds.setValidationQuery("select 1");
+                try {
+                    ds.init();
+                } catch (SQLException e) {
+                    throw new RuntimeException(e.getMessage(), e);
+                }
+                DatasourceConfig.DATA_SOURCES.put(entry.getKey(), ds);
+            }
+        }
+    }
 }

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java

@@ -1,84 +0,0 @@
-package com.alibaba.otter.canal.adapter.launcher.config;
-
-import java.sql.SQLException;
-import java.util.List;
-import java.util.Map;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.stereotype.Component;
-
-import com.alibaba.druid.pool.DruidDataSource;
-import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
-import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
-
-/**
- * 适配器数据源及配置文件列表配置类
- *
- * @author rewerma @ 2018-10-20
- * @version 1.0.0
- */
-@Component
-@ConfigurationProperties(prefix = "adapter.conf")
-public class AdapterConfig {
-
-    private static Logger                 logger = LoggerFactory.getLogger(AdapterConfig.class);
-
-    private Map<String, DatasourceConfig> datasourceConfigs;
-
-    private List<String>                  adapterConfigs;
-
-    public List<String> getAdapterConfigs() {
-        return adapterConfigs;
-    }
-
-    public Map<String, DatasourceConfig> getDatasourceConfigs() {
-        return datasourceConfigs;
-    }
-
-    public void setDatasourceConfigs(Map<String, DatasourceConfig> datasourceConfigs) {
-        this.datasourceConfigs = datasourceConfigs;
-
-        if (datasourceConfigs != null) {
-            for (Map.Entry<String, DatasourceConfig> entry : datasourceConfigs.entrySet()) {
-                DatasourceConfig datasourceConfig = entry.getValue();
-                // 加载数据源连接池
-                DruidDataSource ds = new DruidDataSource();
-                ds.setDriverClassName(datasourceConfig.getDriver());
-                ds.setUrl(datasourceConfig.getUrl());
-                ds.setUsername(datasourceConfig.getUsername());
-                ds.setPassword(datasourceConfig.getPassword());
-                ds.setInitialSize(1);
-                ds.setMinIdle(1);
-                ds.setMaxActive(datasourceConfig.getMaxActive());
-                ds.setMaxWait(60000);
-                ds.setTimeBetweenEvictionRunsMillis(60000);
-                ds.setMinEvictableIdleTimeMillis(300000);
-                ds.setValidationQuery("select 1");
-                try {
-                    ds.init();
-                } catch (SQLException e) {
-                    logger.error("ERROR ## failed to initial datasource: " + datasourceConfig.getUrl(), e);
-                }
-                DatasourceConfig.DATA_SOURCES.put(entry.getKey(), ds);
-            }
-        }
-    }
-
-    public void setAdapterConfigs(List<String> adapterConfigs) {
-        this.adapterConfigs = adapterConfigs;
-
-        if (adapterConfigs != null) {
-            AdapterConfigs.clear();
-            for (String adapterConfig : adapterConfigs) {
-                int idx = adapterConfig.indexOf("/");
-                if (idx > -1) {
-                    String type = adapterConfig.substring(0, idx);
-                    String ymlFile = adapterConfig.substring(idx + 1);
-                    AdapterConfigs.put(type, ymlFile);
-                }
-            }
-        }
-    }
-}

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java

@@ -33,6 +33,8 @@ public class CanalAdapterLoader {
 
     private Map<String, AbstractCanalAdapterWorker> canalMQWorker = new HashMap<>();
 
+    private Map<String, OuterAdapter>               outerAdapters = new HashMap<>();                                  // 配置文件对应adapter
+
     private ExtensionLoader<OuterAdapter>           loader;
 
     public CanalAdapterLoader(CanalClientConfig canalClientConfig){
@@ -53,8 +55,8 @@ public class CanalAdapterLoader {
         }
         String zkHosts = this.canalClientConfig.getZookeeperHosts();
 
-        // 初始化canal-client的适配器
         if (canalClientConfig.getCanalInstances() != null) {
+            // 初始化canal-client的适配器
             for (CanalClientConfig.CanalInstance instance : canalClientConfig.getCanalInstances()) {
                 List<List<OuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();
 
@@ -83,10 +85,8 @@ public class CanalAdapterLoader {
                 worker.start();
                 logger.info("Start adapter for canal instance: {} succeed", instance.getInstance());
             }
-        }
+        } else if (canalClientConfig.getMqTopics() != null) {
-
+            // 初始化canal-client-mq的适配器
-        // 初始化canal-client-mq的适配器
-        if (canalClientConfig.getMqTopics() != null) {
             for (CanalClientConfig.MQTopic topic : canalClientConfig.getMqTopics()) {
                 for (CanalClientConfig.MQGroup group : topic.getGroups()) {
                     List<List<OuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();
@@ -124,11 +124,11 @@ public class CanalAdapterLoader {
     private void loadConnector(OuterAdapterConfig config, List<OuterAdapter> canalOutConnectors) {
         try {
             OuterAdapter adapter;
-            if ("rdb".equalsIgnoreCase(config.getName())) {
+            // if ("rdb".equalsIgnoreCase(config.getName())) {
-                adapter = loader.getExtension(config.getName(), config.getKey());
+            adapter = loader.getExtension(config.getName(), StringUtils.trimToEmpty(config.getKey()));
-            } else {
+            // } else {
-                adapter = loader.getExtension(config.getName());
+            // adapter = loader.getExtension(config.getName());
-            }
+            // }
             ClassLoader cl = Thread.currentThread().getContextClassLoader();
             // 替换ClassLoader
             Thread.currentThread().setContextClassLoader(adapter.getClass().getClassLoader());

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java

@@ -12,7 +12,6 @@ import org.springframework.stereotype.Component;
 
 import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.otter.canal.adapter.launcher.config.AdapterCanalConfig;
-import com.alibaba.otter.canal.adapter.launcher.config.AdapterConfig;
 import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 
 /**
@@ -33,8 +32,6 @@ public class CanalAdapterService {
 
     // 注入bean保证优先注册
     @Resource
-    private AdapterConfig             adapterConfig;
-    @Resource
     private SpringContext             springContext;
     @Resource
     private SyncSwitch                syncSwitch;

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java

@@ -46,17 +46,17 @@ public class CommonRest {
     }
 
     /**
-     * ETL curl http://127.0.0.1:8081/etl/hbase/mytest_person2.yml -X POST
+     * ETL curl http://127.0.0.1:8081/etl/rdb/oracle1/mytest_user.yml -X POST
-     * 
+     *
      * @param type 类型 hbase, es
-     * @param task 任务名对应配置文件名 mytest_person2.yml
+     * @param key adapter key
+     * @param task 任务名对应配置文件名 mytest_user.yml
      * @param params etl where条件参数, 为空全部导入
-     * @return
      */
-    @PostMapping("/etl/{type}/{task}")
+    @PostMapping("/etl/{type}/{key}/{task}")
-    public EtlResult etl(@PathVariable String type, @PathVariable String task,
+    public EtlResult etl(@PathVariable String type, @PathVariable String key, @PathVariable String task,
                          @RequestParam(name = "params", required = false) String params) {
-        OuterAdapter adapter = loader.getExtension(type);
+        OuterAdapter adapter = loader.getExtension(type, key);
         String destination = adapter.getDestination(task);
         String lockKey = destination == null ? task : destination;
 
@@ -83,12 +83,11 @@ public class CommonRest {
                 }
             }
             try {
-                List<String> paramArr = null;
+                List<String> paramArray = null;
                 if (params != null) {
-                    String[] parmaArray = params.trim().split(";");
+                    paramArray = Arrays.asList(params.trim().split(";"));
-                    paramArr = Arrays.asList(parmaArray);
                 }
-                return adapter.etl(task, paramArr);
+                return adapter.etl(task, paramArray);
             } finally {
                 if (destination != null && oriSwitchStatus != null && oriSwitchStatus) {
                     syncSwitch.on(destination);
@@ -101,6 +100,33 @@ public class CommonRest {
         }
     }
 
+    /**
+     * ETL curl http://127.0.0.1:8081/etl/hbase/mytest_person2.yml -X POST
+     * 
+     * @param type 类型 hbase, es
+     * @param task 任务名对应配置文件名 mytest_person2.yml
+     * @param params etl where条件参数, 为空全部导入
+     */
+    @PostMapping("/etl/{type}/{task}")
+    public EtlResult etl(@PathVariable String type, @PathVariable String task,
+                         @RequestParam(name = "params", required = false) String params) {
+        return etl(type, null, task, params);
+    }
+
+    /**
+     * 统计总数 curl http://127.0.0.1:8081/count/rdb/oracle1/mytest_user.yml
+     *
+     * @param type 类型 hbase, es
+     * @param key adapter key
+     * @param task 任务名对应配置文件名 mytest_person2.yml
+     * @return
+     */
+    @GetMapping("/count/{type}/{key}/{task}")
+    public Map<String, Object> count(@PathVariable String type, @PathVariable String key, @PathVariable String task) {
+        OuterAdapter adapter = loader.getExtension(type, key);
+        return adapter.count(task);
+    }
+
     /**
      * 统计总数 curl http://127.0.0.1:8081/count/hbase/mytest_person2.yml
      * 
@@ -110,8 +136,7 @@ public class CommonRest {
      */
     @GetMapping("/count/{type}/{task}")
     public Map<String, Object> count(@PathVariable String type, @PathVariable String task) {
-        OuterAdapter adapter = loader.getExtension(type);
+        return count(type, null, task);
-        return adapter.count(task);
     }
 
     /**

client-adapter/launcher/src/main/resources/application.yml

@@ -16,6 +16,11 @@ canal.conf:
 #  zookeeperHosts: slave1:2181
 #  bootstrapServers: slave1:6667 #or rocketmq
 #  flatMessage: true
+#  srcDataSources:
+#    defaultDS:
+#      url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true
+#      username: root
+#      password: 121212
   canalInstances:
   - instance: example
     groups:
@@ -25,7 +30,7 @@ canal.conf:
 #        key: oracle1
 #        properties:
 #          jdbc.driverClassName: oracle.jdbc.OracleDriver
-#          jdbc.url: jdbc:oracle:thin:@127.0.0.1:1521:orcl
+#          jdbc.url: jdbc:oracle:thin:@localhost:49161:XE
 #          jdbc.username: mytest
 #          jdbc.password: m121212
 #      - name: rdb
@@ -45,27 +50,9 @@ canal.conf:
 #        properties:
 #          cluster.name: elasticsearch
 #  mqTopics:
-#  - mqMode: kafka
+#  - mqMode: kafka # or rocketmq
 #    topic: example
 #    groups:
 #    - groupId: g2
 #      outAdapters:
-#      - name: logger
+#      - name: logger
-#  mqTopics:
-#  - mqMode: rocketmq
-#    topic: example
-#    groups:
-#    - groupId: g2
-#      outAdapters:
-#      - name: logger
-
-#adapter.conf:
-#  datasourceConfigs:
-#    defaultDS:
-#      url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true
-#      username: root
-#      password: 121212
-#  adapterConfigs:
-#  - hbase/mytest_person2.yml
-#  - es/mytest_user.yml
-#  - rdb/mytest_user.yml

client-adapter/rdb/README.md

@@ -0,0 +1,67 @@
+## RDB适配器
+RDB adapter 用于适配mysql到任意关系型数据库(需支持jdbc)的数据同步及导入
+### 1.1 修改启动器配置: application.yml, 这里以oracle目标库为例
+```
+server:
+  port: 8081
+logging:
+  level:
+    com.alibaba.otter.canal.client.adapter.hbase: DEBUG
+spring:
+  jackson:
+    date-format: yyyy-MM-dd HH:mm:ss
+    time-zone: GMT+8
+    default-property-inclusion: non_null
+
+canal.conf:
+  canalServerHost: 127.0.0.1:11111
+  srcDataSources:
+    defaultDS:
+      url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true
+      username: root
+      password: 121212
+  canalInstances:
+  - instance: example
+    groups:
+    - outAdapters:
+      - name: rdb
+        key: oracle1
+        properties:
+          jdbc.driverClassName: oracle.jdbc.OracleDriver
+          jdbc.url: jdbc:oracle:thin:@localhost:49161:XE
+          jdbc.username: mytest
+          jdbc.password: m121212
+```
+其中 outAdapter 的配置: name统一为rdb, key为对应的数据源的唯一标识需和下面的表映射文件中的outerAdapterKey对应!! properties为目标库jdb的相关参数
+adapter将会自动加载 conf/rdb 下的所有.yml结尾的表映射配置文件
+### 1.2 适配器表映射文件
+修改 conf/rdb/mytest_user.yml文件:
+```
+dataSourceKey: defaultDS        # 源数据源的key, 对应上面配置的srcDataSources中的值
+destination: example            # cannal的instance或者MQ的topic
+outerAdapterKey: oracle1        # adapter key, 对应上面配置outAdapters中的key
+dbMapping:
+  database: mytest              # 源数据源的database/shcema
+  table: user                   # 源数据源表名
+  targetTable: mytest.tb_user   # 目标数据源的库名.表名
+  targetPk:                     # 主键映射
+    id: id                      # 如果是复合主键可以换行映射多个
+  mapAll: true                  # 是否整表映射, 要求源表和目标表字段名一模一样 (如果targetColumns也配置了映射,则以targetColumns配置为准)
+#  targetColumns:               # 字段映射, 格式: 目标表字段: 源表字段, 如果字段名一样源表字段名可不填
+#    id:
+#    name:
+#    role_id:
+#    c_time:
+#    test1: 
+```
+导入的类型以目标表的元类型为准, 将自动转换
+
+### 1.3 启动RDB数据同步
+#### 将目标库的jdbc jar包放入lib文件夹, 这里放入ojdbc6.jar
+
+#### 启动canal-adapter启动器
+```
+bin/startup.sh
+```
+#### 验证
+修改mysql mytest.user表的数据, 将会自动同步到Oracle的MYTEST.TB_USER表下面, 并会打出DML的log

client-adapter/rdb/pom.xml

@@ -80,6 +80,26 @@
                     </execution>
                 </executions>
             </plugin>
+            <plugin>
+                <artifactId>maven-antrun-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>run</goal>
+                        </goals>
+                        <configuration>
+                            <tasks>
+                                <copy todir="${project.basedir}/../launcher/target/classes/rdb" overwrite="true" >
+                                    <fileset dir="${project.basedir}/target/classes/rdb" erroronmissingdir="true">
+                                        <include name="*.yml"/>
+                                    </fileset>
+                                </copy>
+                            </tasks>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
         </plugins>
     </build>
 </project>

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java

@@ -1,9 +1,11 @@
 package com.alibaba.otter.canal.client.adapter.rdb;
 
 import java.sql.Connection;
-import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
-import java.util.*;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
 
 import javax.sql.DataSource;
 
@@ -13,8 +15,8 @@ import org.slf4j.LoggerFactory;
 
 import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.otter.canal.client.adapter.OuterAdapter;
+import com.alibaba.otter.canal.client.adapter.rdb.config.ConfigLoader;
 import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
-import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfigLoader;
 import com.alibaba.otter.canal.client.adapter.rdb.service.RdbEtlService;
 import com.alibaba.otter.canal.client.adapter.rdb.service.RdbSyncService;
 import com.alibaba.otter.canal.client.adapter.support.*;
@@ -22,26 +24,26 @@ import com.alibaba.otter.canal.client.adapter.support.*;
 @SPI("rdb")
 public class RdbAdapter implements OuterAdapter {
 
-    private static Logger                       logger     = LoggerFactory.getLogger(RdbAdapter.class);
+    private static Logger              logger             = LoggerFactory.getLogger(RdbAdapter.class);
 
-    private volatile Map<String, MappingConfig> rdbMapping = new HashMap<>();                          // 文件名对应配置
+    private Map<String, MappingConfig> rdbMapping         = new HashMap<>();                          // 文件名对应配置
-    private Map<String, MappingConfig>          mappingConfigCache;                                    // 库名-表名对应配置
+    private Map<String, MappingConfig> mappingConfigCache = new HashMap<>();                          // 库名-表名对应配置
 
-    private DruidDataSource                     dataSource;
+    private DruidDataSource            dataSource;
 
-    private RdbSyncService                      rdbSyncService;
+    private RdbSyncService             rdbSyncService;
 
     @Override
     public void init(OuterAdapterConfig configuration) {
-        SPI spi = this.getClass().getAnnotation(SPI.class);
+        Map<String, MappingConfig> rdbMappingTmp = ConfigLoader.load();
-        Map<String, MappingConfig> rdbMappingTmp = MappingConfigLoader.load(spi.value());
+        // 过滤不匹配的key的配置
-        // 过滤其他key的配置
         rdbMappingTmp.forEach((key, mappingConfig) -> {
-            if (mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey())) {
+            if ((mappingConfig.getOuterAdapterKey() == null && configuration.getKey() == null)
+                || (mappingConfig.getOuterAdapterKey() != null
+                    && mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) {
                 rdbMapping.put(key, mappingConfig);
             }
         });
-        mappingConfigCache = new HashMap<>();
         for (MappingConfig mappingConfig : rdbMapping.values()) {
             mappingConfigCache
                 .put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "."
@@ -68,55 +70,6 @@ public class RdbAdapter implements OuterAdapter {
             logger.error("ERROR ## failed to initial datasource: " + properties.get("jdbc.url"), e);
         }
 
-        rdbMapping.values().forEach(config -> {
-            try {
-                MappingConfig.DbMapping dbMapping = config.getDbMapping();
-                // 从源表加载所有字段名
-                if (dbMapping.getAllColumns() == null) {
-                    synchronized (RdbSyncService.class) {
-                        if (dbMapping.getAllColumns() == null) {
-                            DataSource srcDS = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
-                            Connection srcConn = srcDS.getConnection();
-                            String srcMetaSql = "SELECT * FROM " + dbMapping.getDatabase() + "." + dbMapping.getTable()
-                                                + " WHERE 1=2 ";
-                            List<String> srcColumns = new ArrayList<>();
-                            Util.sqlRS(srcConn, srcMetaSql, rs -> {
-                                try {
-                                    ResultSetMetaData rmd = rs.getMetaData();
-                                    int cnt = rmd.getColumnCount();
-                                    for (int i = 1; i <= cnt; i++) {
-                                        srcColumns.add(rmd.getColumnName(i).toLowerCase());
-                                    }
-                                } catch (SQLException e) {
-                                    logger.error(e.getMessage(), e);
-                                }
-                            });
-                            Map<String, String> columnsMap = new LinkedHashMap<>();
-
-                            for (String srcColumn : srcColumns) {
-                                String targetColumn = srcColumn;
-                                if (dbMapping.getTargetColumns() != null) {
-                                    for (Map.Entry<String, String> entry : dbMapping.getTargetColumns().entrySet()) {
-                                        String targetColumnName = entry.getKey();
-                                        String srcColumnName = entry.getValue();
-
-                                        if (srcColumnName != null
-                                            && srcColumnName.toLowerCase().equals(srcColumn.toUpperCase())) {
-                                            targetColumn = targetColumnName;
-                                        }
-                                    }
-                                }
-                                columnsMap.put(targetColumn, srcColumn);
-                            }
-                            dbMapping.setAllColumns(columnsMap);
-                        }
-                    }
-                }
-            } catch (SQLException e) {
-                logger.error(e.getMessage(), e);
-            }
-        });
-
         rdbSyncService = new RdbSyncService(dataSource);
     }
 
@@ -213,6 +166,15 @@ public class RdbAdapter implements OuterAdapter {
         return res;
     }
 
+    @Override
+    public String getDestination(String task) {
+        MappingConfig config = rdbMapping.get(task);
+        if (config != null) {
+            return config.getDestination();
+        }
+        return null;
+    }
+
     @Override
     public void destroy() {
         if (dataSource != null) {

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java

@@ -0,0 +1,46 @@
+package com.alibaba.otter.canal.client.adapter.rdb.config;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.yaml.snakeyaml.Yaml;
+
+import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
+
+/**
+ * RDB表映射配置加载器
+ *
+ * @author rewerma 2018-11-07 下午02:41:34
+ * @version 1.0.0
+ */
+public class ConfigLoader {
+
+    private static Logger logger = LoggerFactory.getLogger(ConfigLoader.class);
+
+    /**
+     * 加载HBase表映射配置
+     * 
+     * @return 配置名/配置文件名--对象
+     */
+    public static Map<String, MappingConfig> load() {
+        logger.info("## Start loading rdb mapping config ... ");
+
+        Map<String, MappingConfig> result = new LinkedHashMap<>();
+
+        Map<String, String> configContentMap = MappingConfigsLoader.loadConfigs("rdb");
+        configContentMap.forEach((fileName, content) -> {
+            MappingConfig config = new Yaml().loadAs(content, MappingConfig.class);
+            try {
+                config.validate();
+            } catch (Exception e) {
+                throw new RuntimeException("ERROR Config: " + fileName + " " + e.getMessage(), e);
+            }
+            result.put(fileName, config);
+        });
+
+        logger.info("## Rdb mapping config loaded");
+        return result;
+    }
+}

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java

@@ -79,7 +79,7 @@ public class MappingConfig {
         private int                          readBatch   = 5000;
         private int                          commitBatch = 5000;                  // etl等批量提交大小
 
-        private volatile Map<String, String> allColumns;                          // mapAll为true,自动设置改字段
+//        private volatile Map<String, String> allColumns;                          // mapAll为true,自动设置改字段
 
         public String getDatabase() {
             return database;
@@ -161,12 +161,12 @@ public class MappingConfig {
             this.commitBatch = commitBatch;
         }
 
-        public Map<String, String> getAllColumns() {
+//        public Map<String, String> getAllColumns() {
-            return allColumns;
+//            return allColumns;
-        }
+//        }
-
+//
-        public void setAllColumns(Map<String, String> allColumns) {
+//        public void setAllColumns(Map<String, String> allColumns) {
-            this.allColumns = allColumns;
+//            this.allColumns = allColumns;
-        }
+//        }
     }
 }

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java

@@ -1,100 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.rdb.config;
-
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.charset.StandardCharsets;
-import java.util.Collection;
-import java.util.LinkedHashMap;
-import java.util.Map;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.yaml.snakeyaml.Yaml;
-
-import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
-
-/**
- * RDB表映射配置加载器
- *
- * @author rewerma 2018-11-07 下午02:41:34
- * @version 1.0.0
- */
-public class MappingConfigLoader {
-
-    private static Logger       logger    = LoggerFactory.getLogger(MappingConfigLoader.class);
-
-    /**
-     * 加载HBase表映射配置
-     * 
-     * @return 配置名/配置文件名--对象
-     */
-    public static Map<String, MappingConfig> load(String name) {
-        logger.info("## Start loading rdb mapping config ... ");
-
-        Map<String, MappingConfig> result = new LinkedHashMap<>();
-
-        Collection<String> configs = AdapterConfigs.get(name);
-        if (configs == null) {
-            return result;
-        }
-        for (String c : configs) {
-            if (c == null) {
-                continue;
-            }
-            c = c.trim();
-            if (c.equals("") || c.startsWith("#")) {
-                continue;
-            }
-
-            String configContent = null;
-
-            if (c.endsWith(".yml")) {
-                configContent = readConfigContent(name + "/" + c);
-            }
-
-            MappingConfig config = new Yaml().loadAs(configContent, MappingConfig.class);
-
-            try {
-                config.validate();
-            } catch (Exception e) {
-                throw new RuntimeException("ERROR Config: " + c + " " + e.getMessage(), e);
-            }
-            result.put(c, config);
-        }
-
-        logger.info("## Rdb mapping config loaded");
-        return result;
-    }
-
-    public static String readConfigContent(String config) {
-        InputStream in = null;
-        try {
-            // 先取本地文件，再取类路径
-            File configFile = new File("../conf/" + config);
-            if (configFile.exists()) {
-                in = new FileInputStream(configFile);
-            } else {
-                in = MappingConfigLoader.class.getClassLoader().getResourceAsStream(config);
-            }
-            if (in == null) {
-                throw new RuntimeException("Rdb mapping config file not found.");
-            }
-
-            byte[] bytes = new byte[in.available()];
-            in.read(bytes);
-            return new String(bytes, StandardCharsets.UTF_8);
-        } catch (IOException e) {
-            throw new RuntimeException("Read rdb mapping config  error. ", e);
-        } finally {
-            try {
-                if (in != null) {
-                    in.close();
-                }
-            } catch (IOException e) {
-                // ignore
-            }
-        }
-    }
-}

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java

@@ -13,6 +13,7 @@ import java.util.concurrent.atomic.AtomicLong;
 
 import javax.sql.DataSource;
 
+import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -91,8 +92,8 @@ public class RdbEtlService {
                     } else {
                         sqlFinal = sql + " LIMIT " + offset + "," + cnt;
                     }
-                    Future<Boolean> future = executor.submit(
+                    Future<Boolean> future = executor
-                        () -> executeSqlImport(params, srcDS, targetDS, sqlFinal, dbMapping, successCount, errMsg));
+                        .submit(() -> executeSqlImport(srcDS, targetDS, sqlFinal, dbMapping, successCount, errMsg));
                     futures.add(future);
                 }
 
@@ -102,7 +103,7 @@ public class RdbEtlService {
 
                 executor.shutdown();
             } else {
-                executeSqlImport(params, srcDS, targetDS, sql.toString(), dbMapping, successCount, errMsg);
+                executeSqlImport(srcDS, targetDS, sql.toString(), dbMapping, successCount, errMsg);
             }
 
             logger.info(
@@ -160,8 +161,8 @@ public class RdbEtlService {
     /**
      * 执行导入
      */
-    private static boolean executeSqlImport(List<String> params, DataSource srcDS, DataSource targetDS, String sql,
+    private static boolean executeSqlImport(DataSource srcDS, DataSource targetDS, String sql, DbMapping dbMapping,
-                                            DbMapping dbMapping, AtomicLong successCount, List<String> errMsg) {
+                                            AtomicLong successCount, List<String> errMsg) {
         try {
             Util.sqlRS(srcDS, sql, rs -> {
                 int idx = 1;
@@ -172,16 +173,18 @@ public class RdbEtlService {
                     Map<String, Integer> columnType = new LinkedHashMap<>();
                     ResultSetMetaData rsd = rs.getMetaData();
                     int columnCount = rsd.getColumnCount();
+                    List<String> columns = new ArrayList<>();
                     for (int i = 1; i <= columnCount; i++) {
                         columnType.put(rsd.getColumnName(i).toLowerCase(), rsd.getColumnType(i));
+                        columns.add(rsd.getColumnName(i));
                     }
 
-                    Map<String, String> columnsMap;
+                    Map<String, String> columnsMap = SyncUtil.getColumnsMap(dbMapping, columns);
-                    if (dbMapping.isMapAll()) {
+                    // if (dbMapping.isMapAll()) {
-                        columnsMap = dbMapping.getAllColumns();
+                    // columnsMap = dbMapping.getAllColumns();
-                    } else {
+                    // } else {
-                        columnsMap = dbMapping.getTargetColumns();
+                    // columnsMap = dbMapping.getTargetColumns();
-                    }
+                    // }
 
                     StringBuilder insertSql = new StringBuilder();
                     insertSql.append("INSERT INTO ").append(dbMapping.getTargetTable()).append(" (");

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java

@@ -10,11 +10,9 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.function.Consumer;
 
 import javax.sql.DataSource;
 
-import com.alibaba.otter.canal.client.adapter.support.Util;
 import org.joda.time.DateTime;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -23,8 +21,9 @@ import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.serializer.SerializerFeature;
 import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
 import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig.DbMapping;
-import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
+import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil;
 import com.alibaba.otter.canal.client.adapter.support.Dml;
+import com.alibaba.otter.canal.client.adapter.support.Util;
 
 /**
  * RDB同步操作业务
@@ -84,12 +83,7 @@ public class RdbSyncService {
         Connection conn = dataSource.getConnection();
         conn.setAutoCommit(false);
         try {
-            Map<String, String> columnsMap;
+            Map<String, String> columnsMap = SyncUtil.getColumnsMap(dbMapping, data.get(0));
-            if (dbMapping.isMapAll()) {
-                columnsMap = dbMapping.getAllColumns();
-            } else {
-                columnsMap = dbMapping.getTargetColumns();
-            }
 
             StringBuilder insertSql = new StringBuilder();
             insertSql.append("INSERT INTO ").append(dbMapping.getTargetTable()).append(" (");
@@ -148,6 +142,7 @@ public class RdbSyncService {
                 conn.commit();
             }
         } catch (Exception e) {
+            logger.error(e.getMessage(), e);
             conn.rollback();
         } finally {
             conn.close();
@@ -180,12 +175,7 @@ public class RdbSyncService {
         conn.setAutoCommit(false);
 
         try {
-            Map<String, String> columnsMap;
+            Map<String, String> columnsMap = SyncUtil.getColumnsMap(dbMapping, data.get(0));
-            if (dbMapping.isMapAll()) {
-                columnsMap = dbMapping.getAllColumns();
-            } else {
-                columnsMap = dbMapping.getTargetColumns();
-            }
 
             Map<String, Integer> ctype = getTargetColumnType(conn, config);
 
@@ -230,6 +220,7 @@ public class RdbSyncService {
                 conn.commit();
             }
         } catch (Exception e) {
+            logger.error(e.getMessage(), e);
             conn.rollback();
         } finally {
             conn.close();
@@ -283,14 +274,13 @@ public class RdbSyncService {
                 conn.commit();
             }
         } catch (Exception e) {
+            logger.error(e.getMessage(), e);
             conn.rollback();
         } finally {
             conn.close();
         }
     }
 
-
-
     /**
      * 获取目标字段类型
      * 
@@ -309,7 +299,7 @@ public class RdbSyncService {
                     columnType = new LinkedHashMap<>();
                     final Map<String, Integer> columnTypeTmp = columnType;
                     String sql = "SELECT * FROM " + dbMapping.getTargetTable() + " WHERE 1=2";
-                   Util.sqlRS(conn, sql, rs -> {
+                    Util.sqlRS(conn, sql, rs -> {
                         try {
                             ResultSetMetaData rsd = rs.getMetaData();
                             int columnCount = rsd.getColumnCount();
@@ -524,4 +514,5 @@ public class RdbSyncService {
             logger.error(e.getMessage(), e);
         }
     }
+
 }

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java

@@ -0,0 +1,39 @@
+package com.alibaba.otter.canal.client.adapter.rdb.support;
+
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
+
+public class SyncUtil {
+
+    public static Map<String, String> getColumnsMap(MappingConfig.DbMapping dbMapping, Map<String, Object> data) {
+        return getColumnsMap(dbMapping, data.keySet());
+    }
+
+    public static Map<String, String> getColumnsMap(MappingConfig.DbMapping dbMapping, Collection<String> columns) {
+        Map<String, String> columnsMap;
+        if (dbMapping.isMapAll()) {
+            columnsMap = new LinkedHashMap<>();
+            for (String srcColumn : columns) {
+                boolean flag = true;
+                if (dbMapping.getTargetColumns() != null) {
+                    for (Map.Entry<String, String> entry : dbMapping.getTargetColumns().entrySet()) {
+                        if (srcColumn.equals(entry.getValue())) {
+                            columnsMap.put(entry.getKey(), srcColumn);
+                            flag = false;
+                            break;
+                        }
+                    }
+                }
+                if (flag) {
+                    columnsMap.put(srcColumn, srcColumn);
+                }
+            }
+        } else {
+            columnsMap = dbMapping.getTargetColumns();
+        }
+        return columnsMap;
+    }
+}

client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/ConfigLoadTest.java

@@ -6,23 +6,21 @@ import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
+import com.alibaba.otter.canal.client.adapter.rdb.config.ConfigLoader;
 import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
-import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfigLoader;
-import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
 import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 
 public class ConfigLoadTest {
 
     @Before
     public void before() {
-        AdapterConfigs.put("oracle", "mytest_user.yml");
         // 加载数据源连接池
         DatasourceConfig.DATA_SOURCES.put("defaultDS", TestConstant.dataSource);
     }
 
     @Test
     public void testLoad() {
-        Map<String, MappingConfig> configMap =  MappingConfigLoader.load("oracle");
+        Map<String, MappingConfig> configMap =  ConfigLoader.load();
 
         Assert.assertFalse(configMap.isEmpty());
     }

client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java

@@ -14,7 +14,7 @@ public class Common {
 
         OuterAdapterConfig outerAdapterConfig = new OuterAdapterConfig();
         outerAdapterConfig.setName("rdb");
-        outerAdapterConfig.setKey("oralce1");
+        //outerAdapterConfig.setKey("oralce1");
         Map<String, String> properties = new HashMap<>();
         properties.put("jdbc.driveClassName", "oracle.jdbc.OracleDriver");
         properties.put("jdbc.url", "jdbc:oracle:thin:@127.0.0.1:49161:XE");

client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java

@@ -6,7 +6,6 @@ import org.junit.Before;
 import org.junit.Test;
 
 import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter;
-import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
 import com.alibaba.otter.canal.client.adapter.support.Dml;
 
 public class OracleSyncTest {
@@ -15,7 +14,6 @@ public class OracleSyncTest {
 
     @Before
     public void init() {
-        AdapterConfigs.put("rdb", "mytest_user.yml");
         rdbAdapter = Common.init();
     }