diff --git a/client-adapter/adapter-connector/pom.xml b/client-adapter/adapter-connector/pom.xml new file mode 100644 index 00000000..16891c35 --- /dev/null +++ b/client-adapter/adapter-connector/pom.xml @@ -0,0 +1,64 @@ + + + + canal.client-adapter + com.alibaba.otter + 1.1.1-SNAPSHOT + + 4.0.0 + com.alibaba.otter + adapter-connector + pom + canal client adapter connector module for otter ${project.version} + + + + com.alibaba.otter + client-adapter.logger + ${project.version} + + + * + * + + + jar-with-dependencies + + + com.alibaba.otter + client-adapter.hbase + ${project.version} + + + * + * + + + jar-with-dependencies + + + + + + + org.apache.maven.plugins + maven-dependency-plugin + 2.10 + + + copy-dependencies-to-canal-client-service + package + + copy-dependencies + + + ${project.basedir}/../adapter-launcher/target/lib + + + + + + + \ No newline at end of file diff --git a/client-adapter/adapter-launcher/pom.xml b/client-adapter/adapter-launcher/pom.xml new file mode 100644 index 00000000..50197d56 --- /dev/null +++ b/client-adapter/adapter-launcher/pom.xml @@ -0,0 +1,91 @@ + + + + canal.client-adapter + com.alibaba.otter + 1.1.1-SNAPSHOT + + 4.0.0 + com.alibaba.otter + adapter-launcher + jar + canal client adapter launcher module for otter ${project.version} + + + + + org.springframework.boot + spring-boot-dependencies + 2.0.1.RELEASE + pom + import + + + + + + + com.alibaba.otter + client-adapter.common + ${project.version} + + + com.alibaba.otter + canal.client + ${canal_version} + + + org.yaml + snakeyaml + 1.19 + + + org.springframework.boot + spring-boot-starter-web + + + org.springframework.boot + spring-boot-configuration-processor + true + + + + + + + org.springframework.boot + spring-boot-maven-plugin + 2.0.1.RELEASE + + + + repackage + + + + + + maven-antrun-plugin + + + package + + run + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java b/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java new file mode 100644 index 00000000..4b0035d8 --- /dev/null +++ b/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java @@ -0,0 +1,56 @@ +package com.alibaba.otter.canal.adapter.launcher; + +import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; +import javax.annotation.Resource; + +import com.alibaba.otter.canal.adapter.launcher.loader.CanalAdapterLoader; +import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.builder.SpringApplicationBuilder; + + +@SpringBootApplication +public class CanalAdapterApplication { + private static final Logger logger = LoggerFactory.getLogger(CanalAdapterApplication.class); + + private static CanalAdapterLoader adapterLoader; + + public static void main(String[] args) { + new SpringApplicationBuilder(CanalAdapterApplication.class).run(args); + } + + @Resource + private CanalClientConfig canalClientConf; + + @PostConstruct + public void init() { + if (adapterLoader == null) { + try { + logger.info("## start the canal client adapters."); + adapterLoader = new CanalAdapterLoader(canalClientConf); + adapterLoader.init(); + logger.info("## the canal client adapters are running now ......"); + } catch (Throwable e) { + logger.error("## something goes wrong when starting up the canal client adapters:", e); + System.exit(0); + } + } + } + + @PreDestroy + public void destroy() { + try { + logger.info("## stop the canal client adapters"); + if (adapterLoader != null) { + adapterLoader.destroy(); + } + } catch (Throwable e) { + logger.warn("## something goes wrong when stopping canal client adapters:", e); + } finally { + logger.info("## canal client adapters are down."); + } + } +} diff --git a/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java b/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java new file mode 100644 index 00000000..5b232c36 --- /dev/null +++ b/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java @@ -0,0 +1,10 @@ +package com.alibaba.otter.canal.adapter.launcher.config; + +import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.stereotype.Component; + +@ConfigurationProperties(prefix = "canal.conf") +@Component +public class AdapterCanalConfig extends CanalClientConfig { +} diff --git a/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfigPath.java b/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfigPath.java new file mode 100644 index 00000000..474d43af --- /dev/null +++ b/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfigPath.java @@ -0,0 +1,35 @@ +package com.alibaba.otter.canal.adapter.launcher.config; + +import java.util.List; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.stereotype.Component; + +import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; + +@Component +@ConfigurationProperties(prefix = "adapter.conf.path") +public class AdapterConfigPath { + + private List adapterConfigs; + + public List getAdapterConfigs() { + return adapterConfigs; + } + + public void setAdapterConfigs(List adapterConfigs) { + this.adapterConfigs = adapterConfigs; + + if (adapterConfigs != null) { + AdapterConfigs.configs.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.configs.put(type, ymlFile); + } + } + } + } +} diff --git a/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java b/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java new file mode 100644 index 00000000..e0912bab --- /dev/null +++ b/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java @@ -0,0 +1,160 @@ +package com.alibaba.otter.canal.adapter.launcher.loader; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.support.MessageUtil; +import com.alibaba.otter.canal.protocol.Message; + +/** + * 适配器工作线程抽象类 + * + * @author machengyuan 2018-8-19 下午11:30:49 + * @version 1.0.0 + */ +public abstract class AbstractCanalAdapterWorker { + + protected final Logger logger = LoggerFactory.getLogger(this.getClass()); + + protected String canalDestination; // canal实例 + protected List> canalOuterAdapters; // 外部适配器 + protected ExecutorService groupInnerExecutorService; // 组内工作线程池 + protected volatile boolean running = false; // 是否运行中 + protected Thread thread = null; + protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() { + + @Override + public void uncaughtException(Thread t, Throwable e) { + logger.error("parse events has an error", e); + } + }; + + protected void writeOut(final Message message) { + List> futures = new ArrayList<>(); + // 组间适配器并行运行 + for (List outerAdapters : canalOuterAdapters) { + final List adapters = outerAdapters; + futures.add(groupInnerExecutorService.submit(new Callable() { + + @Override + public Boolean call() { + try { + // 组内适配器穿行运行，尽量不要配置组内适配器 + for (final CanalOuterAdapter c : adapters) { + long begin = System.currentTimeMillis(); + MessageUtil.parse4Dml(message, c::writeOut); + + if (logger.isDebugEnabled()) { + logger.debug("{} elapsed time: {}", + c.getClass().getName(), + (System.currentTimeMillis() - begin)); + } + } + return true; + } catch (Exception e) { + return false; + } + } + })); + + // 等待所有适配器写入完成 + // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成 + for (Future f : futures) { + try { + if (!f.get()) { + logger.error("Outer adapter write failed"); + } + } catch (InterruptedException | ExecutionException e) { + // ignore + } + } + } + } + + // protected void writeOut(final FlatMessage flatMessage) { + // List> futures = new ArrayList<>(); + // // 组间适配器并行运行 + // for (List outerAdapters : canalOuterAdapters) { + // final List adapters = outerAdapters; + // futures.add(groupInnerExecutorService.submit(new Callable() { + // + // @Override + // public Boolean call() { + // try { + // // 组内适配器穿行运行，尽量不要配置组内适配器 + // for (CanalOuterAdapter c : adapters) { + // long begin = System.currentTimeMillis(); + // Dml dml = MessageUtil.flatMessage2Dml(flatMessage); + // c.writeOut(dml); + // if (logger.isDebugEnabled()) { + // logger.debug("{} elapsed time: {}", + // c.getClass().getName(), + // (System.currentTimeMillis() - begin)); + // } + // } + // return true; + // } catch (Exception e) { + // return false; + // } + // } + // })); + // + // // 等待所有适配器写入完成 + // // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成 + // for (Future f : futures) { + // try { + // if (!f.get()) { + // logger.error("Outer adapter write failed"); + // } + // } catch (InterruptedException | ExecutionException e) { + // // ignore + // } + // } + // } + // } + + protected void writeOut(Message message, String topic) { + if (logger.isDebugEnabled()) { + logger.debug("topic: {} batchId: {} batchSize: {} ", topic, message.getId(), message.getEntries().size()); + } + long begin = System.currentTimeMillis(); + writeOut(message); + long now = System.currentTimeMillis(); + if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) { + logger.error("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin); + } + if (logger.isDebugEnabled()) { + logger.debug("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin); + } + } + + protected void stopOutAdapters() { + if (thread != null) { + try { + thread.join(); + } catch (InterruptedException e) { + // ignore + } + } + groupInnerExecutorService.shutdown(); + logger.info("topic connectors' worker thread dead!"); + for (List outerAdapters : canalOuterAdapters) { + for (CanalOuterAdapter adapter : outerAdapters) { + adapter.destroy(); + } + } + logger.info("topic all connectors destroyed!"); + } + + public abstract void start(); + + public abstract void stop(); +} diff --git a/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java b/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java new file mode 100644 index 00000000..71c9d130 --- /dev/null +++ b/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java @@ -0,0 +1,112 @@ +package com.alibaba.otter.canal.adapter.launcher.loader; + +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; +import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; + +/** + * 外部适配器的加载器 + * + * @version 1.0.0 + */ +public class CanalAdapterLoader { + + private static final Logger logger = LoggerFactory.getLogger(CanalAdapterLoader.class); + + private CanalClientConfig canalClientConfig; + + private Map canalWorkers = new HashMap<>(); + + private ExtensionLoader loader; + + public CanalAdapterLoader(CanalClientConfig canalClientConfig) { + this.canalClientConfig = canalClientConfig; + } + + /** + * 初始化canal-client + */ + public void init() { + // canal instances 和 mq topics 配置不能同时为空 + // if (canalClientConfig.getCanalInstances() == null && canalClientConfig.getMqTopics() == null) { + // throw new RuntimeException("Blank config property: canalInstances or canalMQTopics"); + // } + + loader = ExtensionLoader.getExtensionLoader(CanalOuterAdapter.class); + + String canalServerHost = this.canalClientConfig.getCanalServerHost(); + SocketAddress sa = null; + if (canalServerHost != null) { + String[] ipPort = canalServerHost.split(":"); + sa = new InetSocketAddress(ipPort[0], Integer.parseInt(ipPort[1])); + } + String zkHosts = this.canalClientConfig.getZookeeperHosts(); + + // 初始化canal-client的适配器 + if (canalClientConfig.getCanalInstances() != null) { + for (CanalClientConfig.CanalInstance instance : canalClientConfig.getCanalInstances()) { + List> canalOuterAdapterGroups = new ArrayList<>(); + + for (CanalClientConfig.AdapterGroup connectorGroup : instance.getAdapterGroups()) { + List canalOutConnectors = new ArrayList<>(); + for (CanalOuterAdapterConfiguration c : connectorGroup.getOutAdapters()) { + loadConnector(c, canalOutConnectors); + } + canalOuterAdapterGroups.add(canalOutConnectors); + } + CanalAdapterWorker worker; + if (zkHosts != null) { + worker = new CanalAdapterWorker(instance.getInstance(), zkHosts, canalOuterAdapterGroups); + } else { + worker = new CanalAdapterWorker(instance.getInstance(), sa, canalOuterAdapterGroups); + } + canalWorkers.put(instance.getInstance(), worker); + worker.start(); + logger.info("Start adapter for canal instance: {} succeed", instance.getInstance()); + } + } + } + + private void loadConnector(CanalOuterAdapterConfiguration config, List canalOutConnectors) { + try { + CanalOuterAdapter adapter = loader.getExtension(config.getName()); + ClassLoader cl = Thread.currentThread().getContextClassLoader(); + // 替换ClassLoader + Thread.currentThread().setContextClassLoader(adapter.getClass().getClassLoader()); + adapter.init(config); + Thread.currentThread().setContextClassLoader(cl); + canalOutConnectors.add(adapter); + logger.info("Load canal adapter: {} succeed", config.getName()); + } catch (Exception e) { + logger.error("Load canal adapter: {} failed", config.getName(), e); + } + } + + /** + * 销毁所有适配器 为防止canal实例太多造成销毁阻塞, 并行销毁 + */ + public void destroy() { + if (canalWorkers.size() > 0) { + ExecutorService stopExecutorService = Executors.newFixedThreadPool(canalWorkers.size()); + for (CanalAdapterWorker v : canalWorkers.values()) { + final CanalAdapterWorker caw = v; + stopExecutorService.submit(() -> caw.stop()); + } + stopExecutorService.shutdown(); + } + logger.info("All canal adapters destroyed"); + } +} diff --git a/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java b/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java new file mode 100644 index 00000000..7e090c2d --- /dev/null +++ b/client-adapter/adapter-launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java @@ -0,0 +1,186 @@ +package com.alibaba.otter.canal.adapter.launcher.loader; + +import java.net.SocketAddress; +import java.util.List; +import java.util.concurrent.Executors; + +import com.alibaba.otter.canal.client.CanalConnector; +import com.alibaba.otter.canal.client.CanalConnectors; +import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.impl.ClusterCanalConnector; +import com.alibaba.otter.canal.protocol.Message; + +/** + * 原生canal-server对应的client适配器工作线程 + * + * @author machengyuan 2018-8-19 下午11:30:49 + * @version 1.0.0 + */ +public class CanalAdapterWorker extends AbstractCanalAdapterWorker { + + private static final int BATCH_SIZE = 50; + private static final int SO_TIMEOUT = 0; + + private CanalConnector connector; + + /** + * 单台client适配器worker的构造方法 + * + * @param canalDestination canal实例名 + * @param address canal-server地址 + * @param canalOuterAdapters 外部适配器组 + */ + public CanalAdapterWorker(String canalDestination, SocketAddress address, + List> canalOuterAdapters){ + this.canalOuterAdapters = canalOuterAdapters; + this.canalDestination = canalDestination; + groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); + connector = CanalConnectors.newSingleConnector(address, canalDestination, "", ""); + } + + /** + * HA模式下client适配器worker的构造方法 + * + * @param canalDestination canal实例名 + * @param zookeeperHosts zookeeper地址 + * @param canalOuterAdapters 外部适配器组 + */ + public CanalAdapterWorker(String canalDestination, String zookeeperHosts, + List> canalOuterAdapters){ + this.canalOuterAdapters = canalOuterAdapters; + this.canalDestination = canalDestination; + groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); + connector = CanalConnectors.newClusterConnector(zookeeperHosts, canalDestination, "", ""); + ((ClusterCanalConnector) connector).setSoTimeout(SO_TIMEOUT); + + // super.initSwitcher(canalDestination); + } + + @Override + public void start() { + if (!running) { + thread = new Thread(new Runnable() { + + @Override + public void run() { + process(); + } + }); + thread.setUncaughtExceptionHandler(handler); + thread.start(); + running = true; + } + } + + @Override + public void stop() { + try { + if (!running) { + return; + } + + // if (switcher != null && !switcher.state()) { + // switcher.set(true); + // } + + connector.stopRunning(); + running = false; + + logger.info("destination {} is waiting for adapters' worker thread die!", canalDestination); + if (thread != null) { + try { + thread.join(); + } catch (InterruptedException e) { + // ignore + } + } + groupInnerExecutorService.shutdown(); + logger.info("destination {} adapters' worker thread dead!", canalDestination); + for (List outerAdapters : canalOuterAdapters) { + for (CanalOuterAdapter adapter : outerAdapters) { + adapter.destroy(); + } + } + logger.info("destination {} all adapters destroyed!", canalDestination); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + private void process() { + while (!running) + ; // waiting until running == true + while (running) { + try { + // if (switcher != null) { + // switcher.get(); + // } + logger.info("=============> Start to connect destination: {} <=============", this.canalDestination); + connector.connect(); + logger.info("=============> Start to subscribe destination: {} <=============", this.canalDestination); + connector.subscribe(); + logger.info("=============> Subscribe destination: {} succeed <=============", this.canalDestination); + while (running) { + // try { + // if (switcher != null) { + // switcher.get(); + // } + // } catch (TimeoutException e) { + // break; + // } + + // server配置canal.instance.network.soTimeout(默认: 30s) + // 范围内未与server交互，server将关闭本次socket连接 + Message message = connector.getWithoutAck(BATCH_SIZE); // 获取指定数量的数据 + long batchId = message.getId(); + try { + int size = message.getEntries().size(); + + if (batchId == -1 || size == 0) { + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + // ignore + } + } else { + if (logger.isDebugEnabled()) { + logger.debug("destination: {} batchId: {} batchSize: {} ", + this.canalDestination, + batchId, + size); + } + long begin = System.currentTimeMillis(); + writeOut(message); + long now = System.currentTimeMillis(); + if (logger.isDebugEnabled()) { + logger.debug("destination: {} batchId: {} elapsed time: {} ms", + this.canalDestination, + batchId, + now - begin); + } + } + connector.ack(batchId); // 提交确认 + } catch (Exception e) { + connector.rollback(batchId); // 处理失败, 回滚数据 + throw e; + } + } + + } catch (Exception e) { + logger.error("process error!", e); + } finally { + connector.disconnect(); + logger.info("=============> Disconnect destination: {} <=============", this.canalDestination); + } + + if (running) { // is reconnect + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + // ignore + } + } + + } + } +} diff --git a/client-adapter/adapter-launcher/src/main/resources/application.yml b/client-adapter/adapter-launcher/src/main/resources/application.yml new file mode 100644 index 00000000..bb06e75b --- /dev/null +++ b/client-adapter/adapter-launcher/src/main/resources/application.yml @@ -0,0 +1,21 @@ +server: + port: 8081 + +canal.conf: + canalServerHost: 127.0.0.1:11111 + #zookeeperHosts: slave1:2181 + #bootstrapServers: localhost:9092 #or rocketmq nameservers:host1:9876;host2:9876 + flatMessage: false + + canalInstances: + - instance: example + adapterGroups: + - outAdapters: + - name: logger +# - name: hbase +# hosts: slave1:2181 +# properties: {znodeParent: "/hbase-unsecure"} + +adapter.conf.path: + adapterConfigs: + - hbase/mytest_person2.yml \ No newline at end of file diff --git a/client-adapter/common/pom.xml b/client-adapter/common/pom.xml index 93793d09..a581ec4c 100644 --- a/client-adapter/common/pom.xml +++ b/client-adapter/common/pom.xml @@ -16,7 +16,7 @@ com.alibaba.otter canal.protocol - ${project.version} + ${canal_version} org.slf4j 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 new file mode 100644 index 00000000..13e51af1 --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java @@ -0,0 +1,8 @@ +package com.alibaba.otter.canal.client.adapter.support; + +import com.google.common.collect.ArrayListMultimap; +import com.google.common.collect.Multimap; + +public interface AdapterConfigs { + Multimap configs = ArrayListMultimap.create(); +} diff --git a/client-adapter/hbase/pom.xml b/client-adapter/hbase/pom.xml index 1437ed5d..da2746c6 100644 --- a/client-adapter/hbase/pom.xml +++ b/client-adapter/hbase/pom.xml @@ -22,7 +22,8 @@ org.yaml snakeyaml - 1.17 + 1.19 + provided org.apache.hbase @@ -45,15 +46,23 @@ - maven-jar-plugin + org.apache.maven.plugins + maven-assembly-plugin + 2.4 - - true - - - **/hbase-mapping/** - + + jar-with-dependencies + + + + make-assembly + package + + single + + + maven-antrun-plugin @@ -65,9 +74,8 @@ - - - + + 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 929e9d6a..5922d63e 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 @@ -4,6 +4,7 @@ import java.io.File; import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; +import java.util.Collection; import java.util.LinkedHashMap; import java.util.Map; @@ -12,12 +13,10 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.yaml.snakeyaml.Yaml; +import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; + /** * HBase表映射配置加载器 - *

- * 配置统一从hbase-mapping/configs.conf文件作为入口, 该文件包含所有表映射配置的名称或者文件名列表。 - * 每个对应的表配置可以yml配置文件或者以database.table为配置名的简化形式 - *

* * @author machengyuan 2018-8-21 下午06:45:49 * @version 1.0.0 @@ -26,7 +25,7 @@ public class MappingConfigLoader { private static Logger logger = LoggerFactory.getLogger(MappingConfigLoader.class); - private static final String BASE_PATH = "hbase-mapping/"; + private static final String BASE_PATH = "hbase"; /** * 加载HBase表映射配置 @@ -35,12 +34,11 @@ public class MappingConfigLoader { */ public static Map load() { logger.info("## Start loading mapping config ... "); - String mappingConfigContent = readConfigContent(BASE_PATH + "configs.conf"); Map result = new LinkedHashMap<>(); - String[] configLines = mappingConfigContent.split("\n"); - for (String c : configLines) { + Collection configs = AdapterConfigs.configs.get("hbase"); + for (String c : configs) { if (c == null) { continue; } diff --git a/client-adapter/hbase/src/main/resources/hbase-mapping/configs.conf b/client-adapter/hbase/src/main/resources/hbase-mapping/configs.conf deleted file mode 100644 index ff5b292e..00000000 --- a/client-adapter/hbase/src/main/resources/hbase-mapping/configs.conf +++ /dev/null @@ -1,5 +0,0 @@ -# 详细映射配置 -mytest_person2.yml - -# 简易配置, 只用指定数据库名.表名, 详细配置全部使用默认 -mytest.person \ No newline at end of file diff --git a/client-adapter/hbase/src/main/resources/hbase-mapping/mytest_person2.yml b/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml similarity index 100% rename from client-adapter/hbase/src/main/resources/hbase-mapping/mytest_person2.yml rename to client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml diff --git a/client-adapter/pom.xml b/client-adapter/pom.xml index 792ce283..7007de33 100644 --- a/client-adapter/pom.xml +++ b/client-adapter/pom.xml @@ -2,20 +2,43 @@ - - canal - com.alibaba.otter - 1.1.1-SNAPSHOT - 4.0.0 com.alibaba.otter canal.client-adapter + 1.1.1-SNAPSHOT pom canal client adapter module for otter ${project.version} + + + UTF-8 + true + true + 1.8 + 1.8 + UTF-8 + 1.1.1-SNAPSHOT + + common logger hbase + adapter-launcher + adapter-connector + + + + org.apache.maven.plugins + maven-compiler-plugin + 3.7.0 + + ${java_source_version} + ${java_target_version} + ${file_encoding} + + + + diff --git a/client-launcher/pom.xml b/client-launcher/pom.xml index 8c29047a..78c1f164 100644 --- a/client-launcher/pom.xml +++ b/client-launcher/pom.xml @@ -46,13 +46,6 @@ com.alibaba.otter client-adapter.logger ${project.version} - jar-with-dependencies - - - com.alibaba.otter - client-adapter.hbase - ${project.version} - jar-with-dependencies diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index ea096de8..c14fbfec 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -63,7 +63,7 @@ canal.instance.parser.parallel = true canal.instance.parser.parallelBufferSize = 256 # table meta tsdb info -canal.instance.tsdb.enable=true +canal.instance.tsdb.enable=false canal.instance.tsdb.dir=${canal.file.data.dir:../conf}/${canal.instance.destination:} canal.instance.tsdb.url=jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL; canal.instance.tsdb.dbUsername=canal diff --git a/deployer/src/main/resources/example/instance.properties b/deployer/src/main/resources/example/instance.properties index 81f78b65..27e120f5 100644 --- a/deployer/src/main/resources/example/instance.properties +++ b/deployer/src/main/resources/example/instance.properties @@ -18,7 +18,7 @@ canal.instance.rds.secretkey= canal.instance.rds.instanceId= # table meta tsdb info -canal.instance.tsdb.enable=true +canal.instance.tsdb.enable=false #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb #canal.instance.tsdb.dbUsername=canal #canal.instance.tsdb.dbPassword=canal @@ -30,13 +30,13 @@ canal.instance.tsdb.enable=true #canal.instance.standby.gtid= # username/password -canal.instance.dbUsername=canal -canal.instance.dbPassword=cZozNf1mzW6EQLGO2q9u99619xbZLO0fbua3EX08r4BWNXb8lAt1aHrTEOBttd6UY8Vnuc0easlVXZDdLtt8BQ== +canal.instance.dbUsername=root +canal.instance.dbPassword=121212 canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ== canal.instance.connectionCharset = UTF-8 canal.instance.defaultDatabaseName =test # enable druid Decrypt database password -canal.instance.enableDruid=true +canal.instance.enableDruid=false # table regex canal.instance.filter.regex=.*\\..* diff --git a/deployer/src/main/resources/logback.xml b/deployer/src/main/resources/logback.xml index d27a1d78..034beb33 100644 --- a/deployer/src/main/resources/logback.xml +++ b/deployer/src/main/resources/logback.xml @@ -78,8 +78,8 @@ - - - + + + \ No newline at end of file diff --git a/pom.xml b/pom.xml index dc200785..49ddc369 100644 --- a/pom.xml +++ b/pom.xml @@ -247,7 +247,7 @@ com.alibaba.fastsql fastsql - 2.0.0_preview_630 + 2.0.0_preview_540 com.alibaba