From 2e5b7b89af953beb5693baf50d0685f3786ca873 Mon Sep 17 00:00:00 2001 From: mcy Date: Mon, 12 Nov 2018 17:30:40 +0800 Subject: [PATCH] =?UTF-8?q?=E7=AE=80=E5=8C=96adapter=E9=85=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../adapter/support/AdapterConfigs.java | 41 ------ .../adapter/support/MappingConfigsLoader.java | 46 +++++++ client-adapter/elasticsearch/pom.xml | 20 +++ .../canal/client/adapter/es/ESAdapter.java | 6 +- .../adapter/es/config/ESSyncConfigLoader.java | 80 ++---------- .../adapter/es/test/ConfigLoadTest.java | 6 +- .../es/test/sync/LabelSyncJoinSub2Test.java | 11 +- .../es/test/sync/LabelSyncJoinSubTest.java | 11 +- .../es/test/sync/RoleSyncJoinOne2Test.java | 11 +- .../es/test/sync/RoleSyncJoinOneTest.java | 5 +- .../es/test/sync/UserSyncJoinOneTest.java | 7 +- .../es/test/sync/UserSyncSingleTest.java | 8 +- .../resources/es/mytest_user_join_one.yml | 10 -- .../resources/es/mytest_user_join_one2.yml | 10 -- .../resources/es/mytest_user_join_sub.yml | 11 -- .../resources/es/mytest_user_join_sub2.yml | 11 -- ...ser_single.yml => mytest_user_single.yml_} | 0 client-adapter/hbase/pom.xml | 20 +++ .../client/adapter/hbase/HbaseAdapter.java | 6 +- .../hbase/config/MappingConfigLoader.java | 123 ++---------------- .../launcher/config/AdapterCanalConfig.java | 41 +++++- .../launcher/config/AdapterConfig.java | 84 ------------ .../launcher/loader/CanalAdapterService.java | 3 - .../src/main/resources/application.yml | 29 ++--- client-adapter/rdb/pom.xml | 20 +++ .../canal/client/adapter/rdb/RdbAdapter.java | 13 +- .../adapter/rdb/config/ConfigLoader.java | 46 +++++++ .../rdb/config/MappingConfigLoader.java | 100 -------------- .../adapter/rdb/service/RdbSyncService.java | 7 +- .../adapter/rdb/test/ConfigLoadTest.java | 6 +- .../client/adapter/rdb/test/sync/Common.java | 2 +- .../adapter/rdb/test/sync/OracleSyncTest.java | 2 - .../resources/{oracle => rdb}/mytest_user.yml | 0 33 files changed, 260 insertions(+), 536 deletions(-) delete mode 100644 client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java create mode 100644 client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MappingConfigsLoader.java delete mode 100644 client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one.yml delete mode 100644 client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one2.yml delete mode 100644 client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub.yml delete mode 100644 client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub2.yml rename client-adapter/elasticsearch/src/test/resources/es/{mytest_user_single.yml => mytest_user_single.yml_} (100%) delete mode 100644 client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java create mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java delete mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java rename client-adapter/rdb/src/test/resources/{oracle => rdb}/mytest_user.yml (100%) diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java deleted file mode 100644 index 0ccfb509..00000000 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java +++ /dev/null @@ -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> configs = new HashMap<>(); - - public static void put(String key, String value) { - Set values = configs.get(key); - if (values == null) { - values = new LinkedHashSet<>(); - } - values.add(value); - configs.put(key, values); - } - - public static Set get(String key) { - return configs.get(key); - } - - public static void clear() { - configs.clear(); - } -} diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MappingConfigsLoader.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MappingConfigsLoader.java new file mode 100644 index 00000000..711a52b6 --- /dev/null +++ b/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 loadConfigs(String name) { + Map 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; + } +} diff --git a/client-adapter/elasticsearch/pom.xml b/client-adapter/elasticsearch/pom.xml index bff1bffd..2d6c3def 100644 --- a/client-adapter/elasticsearch/pom.xml +++ b/client-adapter/elasticsearch/pom.xml @@ -71,6 +71,26 @@ + + maven-antrun-plugin + + + package + + run + + + + + + + + + + + + + diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java index 702fd042..24cc9029 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java @@ -63,12 +63,12 @@ public class ESAdapter implements OuterAdapter { @Override public void init(OuterAdapterConfig configuration) { try { - SPI spi = this.getClass().getAnnotation(SPI.class); - Map esSyncConfigTmp = ESSyncConfigLoader.load(spi.value()); + Map esSyncConfigTmp = ESSyncConfigLoader.load(); // 过滤不匹配的key的配置 esSyncConfigTmp.forEach((key, config) -> { if ((config.getOuterAdapterKey() == null && configuration.getKey() == null) - || config.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey())) { + || (config.getOuterAdapterKey() != null + && config.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) { esSyncConfig.put(key, config); } }); diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java index 5d43572d..e83a96e6 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java +++ b/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.AdapterConfigs; -import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; +import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader; /** * ES 配置装载器 @@ -29,75 +17,25 @@ import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; */ public class ESSyncConfigLoader { - private static Logger logger = LoggerFactory - .getLogger(ESSyncConfigLoader.class); + private static Logger logger = LoggerFactory.getLogger(ESSyncConfigLoader.class); - public static synchronized Map load(String name) { + public static synchronized Map load() { logger.info("## Start loading es mapping config ... "); Map esSyncConfig = new LinkedHashMap<>(); - Collection configs = AdapterConfigs.get("es"); - if (configs == null) { - return esSyncConfig; - } - for (String c : configs) { - if (c == null) { - continue; - } - c = c.trim(); - if (c.equals("") || c.startsWith("#")) { - continue; - } - - ESSyncConfig config; - String configContent = null; - - if (c.endsWith(".yml")) { - configContent = readConfigContent(name + "/" + c); - } - - config = new Yaml().loadAs(configContent, ESSyncConfig.class); - + Map configContentMap = MappingConfigsLoader.loadConfigs("es"); + configContentMap.forEach((fileName, content) -> { + ESSyncConfig config = new Yaml().loadAs(content, ESSyncConfig.class); try { config.validate(); } 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 - } - } - } } diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ConfigLoadTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ConfigLoadTest.java index 680a3725..7ab197d1 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ConfigLoadTest.java +++ b/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,21 +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() { - Map configMap = ESSyncConfigLoader.load("es"); + Map configMap = ESSyncConfigLoader.load(); ESSyncConfig config = configMap.get("mytest_user.yml"); Assert.assertNotNull(config); Assert.assertEquals("defaultDS", config.getDataSourceKey()); diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java index 5e269a53..2d59f565 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java @@ -2,28 +2,25 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync; import java.util.*; -import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; +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(); } diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java index 411e3066..eafcfe75 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java @@ -2,28 +2,25 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync; import java.util.*; -import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; +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(); } diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java index 94e52242..c87669ba 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java @@ -2,28 +2,25 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync; import java.util.*; -import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; +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(); } diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java index e68b9812..a7a61d5a 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java @@ -4,14 +4,13 @@ import java.util.*; import javax.sql.DataSource; -import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import org.elasticsearch.action.get.GetResponse; import org.junit.Assert; 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; @@ -21,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(); } diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java index f8a5341a..532cbc89 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java @@ -4,16 +4,13 @@ import java.util.*; import javax.sql.DataSource; -import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; 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; @@ -23,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(); } diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java index dccf24a3..bcc3b7f6 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java @@ -2,17 +2,13 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync; import java.util.*; -import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; 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.support.DatasourceConfig; +import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; public class UserSyncSingleTest { @@ -21,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(); } diff --git a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one.yml b/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one.yml deleted file mode 100644 index 8e2f4efa..00000000 --- a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one.yml +++ /dev/null @@ -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 \ No newline at end of file diff --git a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one2.yml b/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one2.yml deleted file mode 100644 index 4a44c5e0..00000000 --- a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_one2.yml +++ /dev/null @@ -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 \ No newline at end of file diff --git a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub.yml b/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub.yml deleted file mode 100644 index 82c4e247..00000000 --- a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub.yml +++ /dev/null @@ -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 \ No newline at end of file diff --git a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub2.yml b/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub2.yml deleted file mode 100644 index c8c22c36..00000000 --- a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_join_sub2.yml +++ /dev/null @@ -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 \ No newline at end of file diff --git a/client-adapter/elasticsearch/src/test/resources/es/mytest_user_single.yml b/client-adapter/elasticsearch/src/test/resources/es/mytest_user_single.yml_ similarity index 100% rename from client-adapter/elasticsearch/src/test/resources/es/mytest_user_single.yml rename to client-adapter/elasticsearch/src/test/resources/es/mytest_user_single.yml_ diff --git a/client-adapter/hbase/pom.xml b/client-adapter/hbase/pom.xml index 623d919b..10d08681 100644 --- a/client-adapter/hbase/pom.xml +++ b/client-adapter/hbase/pom.xml @@ -62,6 +62,26 @@ + + maven-antrun-plugin + + + package + + run + + + + + + + + + + + + + diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java index 7c8d82db..a64ca3dc 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java @@ -47,12 +47,12 @@ public class HbaseAdapter implements OuterAdapter { @Override public void init(OuterAdapterConfig configuration) { try { - SPI spi = this.getClass().getAnnotation(SPI.class); - Map hbaseMappingTmp = MappingConfigLoader.load(spi.value()); + Map hbaseMappingTmp = MappingConfigLoader.load(); // 过滤不匹配的key的配置 hbaseMappingTmp.forEach((key, mappingConfig) -> { if ((mappingConfig.getOuterAdapterKey() == null && configuration.getKey() == null) - || mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey())) { + || (mappingConfig.getOuterAdapterKey() != null + && mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) { hbaseMapping.put(key, mappingConfig); } }); diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java index 058a60f1..1f8c05c1 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java +++ b/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,128 +17,30 @@ 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); /** * 加载HBase表映射配置 * * @return 配置名/配置文件名--对象 */ - public static Map load(String name) { + public static Map load() { logger.info("## Start loading hbase mapping config ... "); Map result = new LinkedHashMap<>(); - Collection 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; - } - - MappingConfig config; - String configContent = null; - - if (c.endsWith(".yml")) { - configContent = readConfigContent(name + "/" + 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); - } - + Map configContentMap = MappingConfigsLoader.loadConfigs("hbase"); + configContentMap.forEach((fileName, content) -> { + MappingConfig config = new Yaml().loadAs(content, 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 - } - } - } } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java index 2579e1f7..9e4737d6 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java +++ b/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 DESTINATIONS = new LinkedHashSet<>(); + public final Set DESTINATIONS = new LinkedHashSet<>(); + + private Map srcDataSources; @Override public void setCanalInstances(List canalInstances) { @@ -48,4 +54,37 @@ public class AdapterCanalConfig extends CanalClientConfig { } } } + + public Map getSrcDataSources() { + return srcDataSources; + } + + public void setSrcDataSources(Map srcDataSources) { + this.srcDataSources = srcDataSources; + + if (srcDataSources != null) { + for (Map.Entry 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); + } + } + } } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java deleted file mode 100644 index 0f18c996..00000000 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java +++ /dev/null @@ -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 datasourceConfigs; - - private List adapterConfigs; - - public List getAdapterConfigs() { - return adapterConfigs; - } - - public Map getDatasourceConfigs() { - return datasourceConfigs; - } - - public void setDatasourceConfigs(Map datasourceConfigs) { - this.datasourceConfigs = datasourceConfigs; - - if (datasourceConfigs != null) { - for (Map.Entry 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 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); - } - } - } - } -} diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java index 2e535535..15dafa79 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java @@ -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; diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 2c48fa84..61866595 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/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 -# 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 +# - name: logger \ No newline at end of file diff --git a/client-adapter/rdb/pom.xml b/client-adapter/rdb/pom.xml index 994d28aa..120aa5e1 100644 --- a/client-adapter/rdb/pom.xml +++ b/client-adapter/rdb/pom.xml @@ -80,6 +80,26 @@ + + maven-antrun-plugin + + + package + + run + + + + + + + + + + + + + \ No newline at end of file diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java index ef1ffe01..185423ee 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java @@ -1,5 +1,6 @@ package com.alibaba.otter.canal.client.adapter.rdb; +import java.io.File; import java.sql.Connection; import java.sql.ResultSetMetaData; import java.sql.SQLException; @@ -14,7 +15,7 @@ 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.MappingConfig; -import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfigLoader; +import com.alibaba.otter.canal.client.adapter.rdb.config.ConfigLoader; 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.*; @@ -33,12 +34,16 @@ public class RdbAdapter implements OuterAdapter { @Override public void init(OuterAdapterConfig configuration) { - SPI spi = this.getClass().getAnnotation(SPI.class); - Map rdbMappingTmp = MappingConfigLoader.load(spi.value()); + System.out.println("xxxxx: " + this.getClass().getClassLoader().getResource("").getPath()); + File file = new File(this.getClass().getClassLoader().getResource("").getPath() + "rdb" + File.separator); + System.out.println(file.getAbsolutePath()); + + Map rdbMappingTmp = ConfigLoader.load(); // 过滤不匹配的key的配置 rdbMappingTmp.forEach((key, mappingConfig) -> { if ((mappingConfig.getOuterAdapterKey() == null && configuration.getKey() == null) - || mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey())) { + || (mappingConfig.getOuterAdapterKey() != null + && mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) { rdbMapping.put(key, mappingConfig); } }); diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java new file mode 100644 index 00000000..c2fe3339 --- /dev/null +++ b/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 load() { + logger.info("## Start loading rdb mapping config ... "); + + Map result = new LinkedHashMap<>(); + + Map 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; + } +} diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java deleted file mode 100644 index aefb467f..00000000 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java +++ /dev/null @@ -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 load(String name) { - logger.info("## Start loading rdb mapping config ... "); - - Map result = new LinkedHashMap<>(); - - Collection 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 - } - } - } -} diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java index ab46bf4b..9a01fbf7 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java @@ -148,6 +148,7 @@ public class RdbSyncService { conn.commit(); } } catch (Exception e) { + logger.error(e.getMessage(), e); conn.rollback(); } finally { conn.close(); @@ -230,6 +231,7 @@ public class RdbSyncService { conn.commit(); } } catch (Exception e) { + logger.error(e.getMessage(), e); conn.rollback(); } finally { conn.close(); @@ -283,14 +285,13 @@ public class RdbSyncService { conn.commit(); } } catch (Exception e) { + logger.error(e.getMessage(), e); conn.rollback(); } finally { conn.close(); } } - - /** * 获取目标字段类型 * @@ -309,7 +310,7 @@ public class RdbSyncService { columnType = new LinkedHashMap<>(); final Map 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(); diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/ConfigLoadTest.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/ConfigLoadTest.java index dea6cee7..5eeb0878 100644 --- a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/ConfigLoadTest.java +++ b/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 configMap = MappingConfigLoader.load("oracle"); + Map configMap = ConfigLoader.load(); Assert.assertFalse(configMap.isEmpty()); } diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java index 4ae04643..5af9fdcb 100644 --- a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java +++ b/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 properties = new HashMap<>(); properties.put("jdbc.driveClassName", "oracle.jdbc.OracleDriver"); properties.put("jdbc.url", "jdbc:oracle:thin:@127.0.0.1:49161:XE"); diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java index 5b8a45e3..a00a9ce5 100644 --- a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java +++ b/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(); } diff --git a/client-adapter/rdb/src/test/resources/oracle/mytest_user.yml b/client-adapter/rdb/src/test/resources/rdb/mytest_user.yml similarity index 100% rename from client-adapter/rdb/src/test/resources/oracle/mytest_user.yml rename to client-adapter/rdb/src/test/resources/rdb/mytest_user.yml