Merge pull request #1042 from rewerma/master

canal adapter readme.md

client-adapter/README.md

@@ -0,0 +1,251 @@
+## 基本说明
+canal 1.1.1版本之后, 增加客户端数据落地的适配及启动功能, 目前支持功能:
+* 客户端启动器
+* 同步管理REST接口
+* 日志适配器, 作为DEMO
+* HBase的数据同步(表对表同步), ETL功能
+* (后续支持) ElasticSearch多表数据同步,ETL功能
+
+## 环境版本
+* 操作系统：无要求
+* java版本: jdk1.8 以上 
+* canal 版本: 请下载最新的安装包，本文以当前v1.1.1 的canal.deployer-1.1.1.tar.gz为例
+* MySQL版本 ：5.7.18
+* HBase版本: Apache HBase 1.1.2, 若和服务端版本不一致可自行替换客户端HBase依赖
+
+## 一、适配器启动器
+client-adapter分为适配器和启动器两部分, 适配器为多个fat jar, 每个适配器会将自己所需的依赖打成一个包, 以SPI的方式让启动器动态加载
+
+
+启动器为 SpringBoot 项目, 支持canal-client启动的同时提供相关REST管理接口, 运行目录结构为:
+```
+canal-adapter-launcher.jar
+- lib
+    client-adapter.logger-1.1.1-jar-with-dependencies.jar
+    client-adapter.hbase-1.1.1-jar-with-dependencies.jar
+- config
+    application.yml
+    - hbase
+        mytest_person2.yml
+```
+以上目录结构最终会打包成 canal-adapter-launcher.tar.gz 压缩包
+
+## 二、启动器
+### 2.1 启动器配置 application.yml
+#### canal相关配置部分说明
+```
+canal.conf:
+  canalServerHost: 127.0.0.1:11111  # 对应单机模式下的canal server的ip:port
+  zookeeperHosts: slave1:2181       # 对应集群模式下的zk地址, 如果配置了canalServerHost, 则以canalServerHost为准
+  bootstrapServers: slave1:6667     # kafka或rocketMQ地址, 与canalServerHost不能并存
+  flatMessage: true                 # 扁平message开关, 是否以json字符串形式投递数据, 仅在kafka/rocketMQ模式下有效
+  canalInstances:                   # canal实例组, 如果是tcp模式可配置此项
+  - instance: example               # 对应canal destination
+    adapterGroups:                  # 对应适配器分组, 分组间的适配器并行运行
+    - outAdapters:                  # 适配器列表, 分组内的适配串行运行
+      - name: logger                # 适配器SPI名
+      - name: hbase
+        properties:                 # HBase相关连接参数
+          hbase.zookeeper.quorum: slave1
+          hbase.zookeeper.property.clientPort: 2181
+          zookeeper.znode.parent: /hbase
+  mqTopics:                         # MQ topic租, 如果是kafka或者rockeMQ模式可配置此项, 与canalInstances不能并存
+  - mqMode: kafka                   # MQ的模式: kafak/rocketMQ
+    topic: example                  # MQ topic
+    groups:                         # group组
+    - groupId: g2                   # group id
+      outAdapters:                  # 适配器列表, 以下配置和canalInstances中一样
+      - name: logger                
+```
+#### 适配器相关配置部分说明
+```
+adapter.conf:
+  datasourceConfigs:                # 数据源配置列表, 数据源将在适配器中用于ETL、数据同步回查等使用
+    defaultDS:                      # 数据源 dataSourceKey, 适配器中通过该值获取指定数据源
+      url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true
+      username: root
+      password: 121212
+  adapterConfigs:                   # 适配器内部配置列表
+  - hbase/mytest_person2.yml        # 类型/配置文件名, 这里示例就是对应HBase适配器hbase目录下的mytest_person2.yml文件
+```
+## 2.2 适配器管理REST接口
+#### 2.2.1 查询所有订阅同步的canal destination或MQ topic
+```
+curl http://127.0.0.1:8081/destinations
+```
+#### 2.2.2 数据同步开关
+```
+curl http://127.0.0.1:8081/syncSwitch/example/off -X PUT
+```
+针对 example 这个canal destination/MQ topic 进行开关操作. off代表关闭, 该destination/topic下的同步将阻塞或者断开连接不再接收数据, on代表开启
+
+注: 如果在配置文件中配置了 zookeeperHosts 项, 则会使用分布式锁来控制HA中的数据同步开关, 如果是单机模式则使用本地锁来控制开关
+#### 2.2.3 数据同步开关状态
+```
+curl http://127.0.0.1:8081/syncSwitch/example
+```
+查看指定 canal destination/MQ topic 的数据同步开关状态
+#### 2.2.4 手动ETL
+```
+curl http://127.0.0.1:8081/etl/hbase/mytest_person2.yml -X POST -d "params=2018-10-21 00:00:00"
+```
+导入数据到指定类型的库
+#### 2.2.5 查看相关库总数据
+```
+curl http://127.0.0.1:8081/count/hbase/mytest_person2.yml
+```
+### 2.3 启动canal-adapter示例
+#### 2.3.1 启动canal server (单机模式), 参考: [Canal QuickStart](https://github.com/alibaba/canal/wiki/QuickStart)
+#### 2.3.2 修改config/application.yml为:
+```
+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
+  flatMessage: true
+  canalInstances:
+  - instance: example
+    adapterGroups:
+    - outAdapters:
+      - name: logger
+```
+启动 canal-adapter-launcher.jar
+```
+java -jar canal-adapter-launcher.jar
+```
+
+## 三、HBase适配器
+### 3.1 修改启动器配置: application.yml
+```
+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
+  flatMessage: true
+  canalInstances:
+  - instance: example
+    adapterGroups:
+    - outAdapters:
+      - name: hbase
+        properties:
+          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
+### 3.2 适配器表映射文件
+修改 config/hbase/mytest_person.yml文件:
+```
+dataSourceKey: defaultDS            # 对应application.yml中的datasourceConfigs下的配置
+hbaseOrm:                           # mysql--HBase的单表映射配置
+  mode: STRING                      # HBase中的存储类型, 默认统一存为String, 可选: #PHOENIX  #NATIVE   #STRING 
+                                    # NATIVE: 以java类型为主, PHOENIX: 将类型转换为Phoenix对应的类型
+  destination: example              # 对应 canal destination/MQ topic 名称
+  database: mytest                  # 数据库名/schema名
+  table: person                     # 表名
+  hbaseTable: MYTEST.PERSON         # HBase表名
+  family: CF                        # 默认统一Column Family名称
+  uppercaseQualifier: true          # 字段名转大写, 默认为true
+  commitBatch: 3000                 # 批量提交的大小, ETL中用到
+  #rowKey: id,type                  # 复合字段rowKey不能和columns中的rowKey并存
+                                    # 复合rowKey会以 '|' 分隔
+  columns:                          # 字段映射, 如果不配置将自动映射所有字段, 
+                                    # 并取第一个字段为rowKey, HBase字段名以mysql字段名为主
+    id: ROWKE                       
+    name: CF:NAME
+    email: EMAIL                    # 如果column family为默认CF, 则可以省略
+    type:                           # 如果HBase字段和mysql字段名一致, 则可以省略
+    c_time: 
+    birthday: 
+```
+如果涉及到类型转换,可以如下形式:
+```
+...
+  columns:                         
+    id: ROWKE$STRING                      
+    ...                   
+    type: TYPE$BYTE                          
+    ...
+```
+类型转换涉及到Java类型和Phoenix类型两种, 分别定义如下:
+```
+#Java 类型转换, 对应配置 mode: NATIVE
+$DEFAULT
+$STRING
+$INTEGER
+$LONG
+$SHORT
+$BOOLEAN
+$FLOAT
+$DOUBLE
+$BIGDECIMAL
+$DATE
+$BYTE
+$BYTES
+```
+```
+#Phoenix 类型转换, 对应配置 mode: PHOENIX
+$DEFAULT                  对应PHOENIX里的VARCHAR
+$UNSIGNED_INT             对应PHOENIX里的UNSIGNED_INT           4字节
+$UNSIGNED_LONG            对应PHOENIX里的UNSIGNED_LONG          8字节
+$UNSIGNED_TINYINT         对应PHOENIX里的UNSIGNED_TINYINT       1字节
+$UNSIGNED_SMALLINT        对应PHOENIX里的UNSIGNED_SMALLINT      2字节
+$UNSIGNED_FLOAT           对应PHOENIX里的UNSIGNED_FLOAT         4字节
+$UNSIGNED_DOUBLE          对应PHOENIX里的UNSIGNED_DOUBLE        8字节
+$INTEGER                  对应PHOENIX里的INTEGER                4字节
+$BIGINT                   对应PHOENIX里的BIGINT                 8字节
+$TINYINT                  对应PHOENIX里的TINYINT                1字节
+$SMALLINT                 对应PHOENIX里的SMALLINT               2字节
+$FLOAT                    对应PHOENIX里的FLOAT                  4字节
+$DOUBLE                    对应PHOENIX里的DOUBLE                 8字节
+$BOOLEAN                  对应PHOENIX里的BOOLEAN                1字节
+$TIME                     对应PHOENIX里的TIME                   8字节
+$DATE                     对应PHOENIX里的DATE                   8字节
+$TIMESTAMP                对应PHOENIX里的TIMESTAMP              12字节
+$UNSIGNED_TIME            对应PHOENIX里的UNSIGNED_TIME          8字节
+$UNSIGNED_DATE            对应PHOENIX里的UNSIGNED_DATE          8字节
+$UNSIGNED_TIMESTAMP       对应PHOENIX里的UNSIGNED_TIMESTAMP     12字节
+$VARCHAR                  对应PHOENIX里的VARCHAR                动态长度
+$VARBINARY                对应PHOENIX里的VARBINARY              动态长度
+$DECIMAL                  对应PHOENIX里的DECIMAL                动态长度
+```
+如果不配置将以java对象原生类型默认映射转换
+### 3.3 启动HBase数据同步
+#### 创建HBase表
+在HBase shell中运行:
+```
+create 'MYTEST.PERSON', {NAME=>'CF'}
+```
+#### 启动canal-adapter启动器
+```
+java -jar canal-adapter-launcher.jar
+```
+#### 验证
+修改mysql mytest.person表的数据, 将会自动同步到HBase的MYTEST.PERSON表下面, 并会打出DML的log

client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml

@@ -45,7 +45,7 @@ hbaseOrm:
 # $TINYINT                  对应PHOENIX里的TINYINT                1字节
 # $SMALLINT                 对应PHOENIX里的SMALLINT               2字节
 # $FLOAT                    对应PHOENIX里的FLOAT                  4字节
-# DOUBLE                    对应PHOENIX里的DOUBLE                 8字节
+# $DOUBLE                    对应PHOENIX里的DOUBLE                 8字节
 # $BOOLEAN                  对应PHOENIX里的BOOLEAN                1字节
 # $TIME                     对应PHOENIX里的TIME                   8字节
 # $DATE                     对应PHOENIX里的DATE                   8字节

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

@@ -2,20 +2,19 @@ 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 com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch;
-import com.alibaba.otter.canal.adapter.launcher.config.SpringContext;
-import com.alibaba.otter.canal.client.adapter.support.Dml;
-import com.alibaba.otter.canal.protocol.FlatMessage;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch;
+import com.alibaba.otter.canal.adapter.launcher.config.SpringContext;
 import com.alibaba.otter.canal.client.adapter.OuterAdapter;
+import com.alibaba.otter.canal.client.adapter.support.Dml;
 import com.alibaba.otter.canal.client.adapter.support.MessageUtil;
+import com.alibaba.otter.canal.protocol.FlatMessage;
 import com.alibaba.otter.canal.protocol.Message;
 
 /**
@@ -44,21 +43,21 @@ public abstract class AbstractCanalAdapterWorker {
     protected void writeOut(final Message message) {
         List<Future<Boolean>> futures = new ArrayList<>();
         // 组间适配器并行运行
-        for (List<OuterAdapter> outerAdapters : canalOuterAdapters) {
+        canalOuterAdapters.forEach(outerAdapters -> {
             final List<OuterAdapter> adapters = outerAdapters;
             futures.add(groupInnerExecutorService.submit(() -> {
                 try {
                     // 组内适配器穿行运行，尽量不要配置组内适配器
-                    for (final OuterAdapter c : adapters) {
+                    adapters.forEach(adapter -> {
                         long begin = System.currentTimeMillis();
-                        MessageUtil.parse4Dml(canalDestination, message, c::sync);
+                        MessageUtil.parse4Dml(canalDestination, message, adapter::sync);
 
                         if (logger.isDebugEnabled()) {
                             logger.debug("{} elapsed time: {}",
-                                c.getClass().getName(),
+                                adapter.getClass().getName(),
                                 (System.currentTimeMillis() - begin));
                         }
-                    }
+                    });
                     return true;
                 } catch (Exception e) {
                     return false;
@@ -67,79 +66,94 @@ public abstract class AbstractCanalAdapterWorker {
 
             // 等待所有适配器写入完成
             // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成
-            for (Future<Boolean> f : futures) {
+            futures.forEach(future -> {
                 try {
-                    if (!f.get()) {
+                    if (!future.get()) {
                         logger.error("Outer adapter write failed");
                     }
                 } catch (InterruptedException | ExecutionException e) {
                     // ignore
                 }
-            }
-        }
+            });
+        });
     }
 
     protected void writeOut(final FlatMessage flatMessage) {
         List<Future<Boolean>> futures = new ArrayList<>();
         // 组间适配器并行运行
-        for (List<OuterAdapter> outerAdapters : canalOuterAdapters) {
-            final List<OuterAdapter> adapters = outerAdapters;
-            futures.add(groupInnerExecutorService.submit(new Callable<Boolean>() {
-
-                @Override
-                public Boolean call() {
-                    try {
-                        // 组内适配器穿行运行，尽量不要配置组内适配器
-                        for (OuterAdapter c : adapters) {
-                            long begin = System.currentTimeMillis();
-                            Dml dml = MessageUtil.flatMessage2Dml(canalDestination, flatMessage);
-                            c.sync(dml);
-                            if (logger.isDebugEnabled()) {
-                                logger.debug("{} elapsed time: {}",
-                                    c.getClass().getName(),
-                                    (System.currentTimeMillis() - begin));
-                            }
+        canalOuterAdapters.forEach(outerAdapters -> {
+            futures.add(groupInnerExecutorService.submit(() -> {
+                try {
+                    // 组内适配器穿行运行，尽量不要配置组内适配器
+                    outerAdapters.forEach(adapter -> {
+                        long begin = System.currentTimeMillis();
+                        Dml dml = MessageUtil.flatMessage2Dml(canalDestination, flatMessage);
+                        adapter.sync(dml);
+                        if (logger.isDebugEnabled()) {
+                            logger.debug("{} elapsed time: {}",
+                                adapter.getClass().getName(),
+                                (System.currentTimeMillis() - begin));
                         }
-                        return true;
-                    } catch (Exception e) {
-                        return false;
-                    }
+                    });
+                    return true;
+                } catch (Exception e) {
+                    return false;
                 }
             }));
 
             // 等待所有适配器写入完成
             // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成
-            for (Future<Boolean> f : futures) {
+            futures.forEach(future -> {
                 try {
-                    if (!f.get()) {
+                    if (!future.get()) {
                         logger.error("Outer adapter write failed");
                     }
                 } catch (InterruptedException | ExecutionException e) {
                     // ignore
                 }
-            }
+            });
+        });
+    }
+
+    public void start() {
+        if (!running) {
+            thread = new Thread(this::process);
+            thread.setUncaughtExceptionHandler(handler);
+            thread.start();
+            running = true;
         }
     }
 
-    protected void stopOutAdapters() {
-        if (thread != null) {
-            try {
-                thread.join();
-            } catch (InterruptedException e) {
-                // ignore
+    protected abstract void process();
+
+    public void stop() {
+        try {
+            if (!running) {
+                return;
             }
-        }
-        groupInnerExecutorService.shutdown();
-        logger.info("topic connectors' worker thread dead!");
-        for (List<OuterAdapter> outerAdapters : canalOuterAdapters) {
-            for (OuterAdapter adapter : outerAdapters) {
-                adapter.destroy();
+
+            closeConnection();
+            running = false;
+
+            syncSwitch.release(canalDestination);
+
+            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);
+            canalOuterAdapters.forEach(outerAdapters -> outerAdapters.forEach(OuterAdapter::destroy));
+            logger.info("destination {} all adapters destroyed!", canalDestination);
+        } catch (Exception e) {
+            logger.error(e.getMessage(), e);
         }
-        logger.info("topic all connectors destroyed!");
     }
 
-    public abstract void start();
+    protected abstract void closeConnection();
 
-    public abstract void stop();
 }

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

@@ -37,59 +37,18 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
         this.flatMessage = flatMessage;
         connector = KafkaCanalConnectors.newKafkaConnector(bootstrapServers, topic, null, groupId, flatMessage);
         // connector.setSessionTimeout(1L, TimeUnit.MINUTES);
-
-        // super.initSwitcher(topic);
     }
 
     @Override
-    public void start() {
-        if (!running) {
-            thread = new Thread(() -> process());
-            thread.setUncaughtExceptionHandler(handler);
-            running = true;
-            thread.start();
-        }
+    protected  void closeConnection(){
+        connector.stopRunning();
     }
 
     @Override
-    public void stop() {
-        try {
-            if (!running) {
-                return;
-            }
-
-            connector.stopRunning();
-            running = false;
-
-            // if (switcher != null && !switcher.state()) {
-            // switcher.set(true);
-            // }
-
-            if (thread != null) {
-                try {
-                    thread.join();
-                } catch (InterruptedException e) {
-                    // ignore
-                }
-            }
-            groupInnerExecutorService.shutdown();
-            logger.info("topic {} connectors' worker thread dead!", this.topic);
-            for (List<OuterAdapter> outerAdapters : canalOuterAdapters) {
-                for (OuterAdapter adapter : outerAdapters) {
-                    adapter.destroy();
-                }
-            }
-            logger.info("topic {} all connectors destroyed!", this.topic);
-        } catch (Exception e) {
-            logger.error(e.getMessage(), e);
-        }
-    }
-
-    private void process() {
+    protected void process() {
         while (!running)
             ;
         ExecutorService executor = Executors.newSingleThreadExecutor();
-        // final AtomicBoolean executing = new AtomicBoolean(true);
         while (running) {
             try {
                 logger.info("=============> Start to connect topic: {} <=============", this.topic);
@@ -99,7 +58,7 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
                 logger.info("=============> Subscribe topic: {} succeed <=============", this.topic);
                 while (running) {
                     try {
-                        // switcher.get(); //等待开关开启
+                        syncSwitch.get(canalDestination);
 
                         List<?> messages;
                         if (!flatMessage) {

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

@@ -135,20 +135,18 @@ public class CanalAdapterLoader {
     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);
+            for (CanalAdapterWorker canalAdapterWorker : canalWorkers.values()) {
+                stopExecutorService.submit(canalAdapterWorker::stop);
             }
             stopExecutorService.shutdown();
         }
 
         if (canalMQWorker.size() > 0) {
-            ExecutorService stopMQWokerService = Executors.newFixedThreadPool(canalMQWorker.size());
-            for (AbstractCanalAdapterWorker tmp : canalMQWorker.values()) {
-                final AbstractCanalAdapterWorker worker = tmp;
-                stopMQWokerService.submit(worker::stop);
+            ExecutorService stopMQWorkerService = Executors.newFixedThreadPool(canalMQWorker.size());
+            for (AbstractCanalAdapterWorker canalAdapterMQWorker : canalMQWorker.values()) {
+                stopMQWorkerService.submit(canalAdapterMQWorker::stop);
             }
-            stopMQWokerService.shutdown();
+            stopMQWorkerService.shutdown();
         }
         logger.info("All canal adapters destroyed");
     }

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

@@ -36,38 +36,12 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
     }
 
     @Override
-    public void start() {
-        if (!running) {
-            thread = new Thread(new Runnable() {
-
-                @Override
-                public void run() {
-                    process();
-                }
-            });
-            thread.setUncaughtExceptionHandler(handler);
-            running = true;
-            thread.start();
-        }
+    protected void closeConnection() {
+        connector.stopRunning();
     }
 
     @Override
-    public void stop() {
-        try {
-            if (!running) {
-                return;
-            }
-            connector.stopRunning();
-            running = false;
-            logger.info("Stop topic {} out adapters begin", this.topic);
-            stopOutAdapters();
-            logger.info("Stop topic {} out adapters end", this.topic);
-        } catch (Exception e) {
-            logger.error(e.getMessage(), e);
-        }
-    }
-
-    private void process() {
+    protected void process() {
         while (!running)
             ;
         ExecutorService executor = Executors.newSingleThreadExecutor();
@@ -84,25 +58,27 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
 
                         final Message message = connector.getWithoutAck(1);
                         if (message != null) {
-                            executor.submit(new Runnable() {
-
-                                @Override
-                                public void run() {
-                                    try {
-                                        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);
-                                        }
-                                    } catch (Exception e) {
-                                        logger.error(e.getMessage(), e);
+                            executor.submit(() -> {
+                                try {
+                                    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);
                                     }
-                                    connector.ack(message.getId());
+                                } catch (Exception e) {
+                                    logger.error(e.getMessage(), e);
                                 }
+                                connector.ack(message.getId());
                             });
                         } else {
                             logger.debug("Message is null");

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

@@ -57,49 +57,12 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
     }
 
     @Override
-    public void start() {
-        if (!running) {
-            thread = new Thread(this::process);
-            thread.setUncaughtExceptionHandler(handler);
-            thread.start();
-            running = true;
-        }
+    protected void closeConnection() {
+        connector.stopRunning();
     }
 
     @Override
-    public void stop() {
-        try {
-            if (!running) {
-                return;
-            }
-
-            connector.stopRunning();
-            running = false;
-
-            syncSwitch.release(canalDestination);
-
-            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<OuterAdapter> outerAdapters : canalOuterAdapters) {
-                for (OuterAdapter adapter : outerAdapters) {
-                    adapter.destroy();
-                }
-            }
-            logger.info("destination {} all adapters destroyed!", canalDestination);
-        } catch (Exception e) {
-            logger.error(e.getMessage(), e);
-        }
-    }
-
-    private void process() {
+    protected void process() {
         while (!running)
             ; // waiting until running == true
         while (running) {
@@ -129,11 +92,7 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
                         int size = message.getEntries().size();
 
                         if (batchId == -1 || size == 0) {
-                            try {
-                                Thread.sleep(1000);
-                            } catch (InterruptedException e) {
-                                // ignore
-                            }
+                            Thread.sleep(1000);
                         } else {
                             if (logger.isDebugEnabled()) {
                                 logger.debug("destination: {} batchId: {} batchSize: {} ",
@@ -143,18 +102,18 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
                             }
                             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);
+                                    System.currentTimeMillis() - begin);
                             }
                         }
                         connector.ack(batchId); // 提交确认
                     } catch (Exception e) {
                         connector.rollback(batchId); // 处理失败, 回滚数据
-                        throw e;
+                        logger.error("sync error!", e);
+                        Thread.sleep(500);
                     }
                 }