fixed issue #1826 , remove kafka transaction send

当按注解散列时，如果监控的表未显示设置主键，将会导致flatMessage.getPkNames()为空，导致无限空指针异常

.codecov.yml

@@ -0,0 +1,6 @@
+coverage:
+  status:
+    # pull-requests only
+    patch:
+      default:
+        threshold: 0.1%

.github/ISSUE_TEMPLATE/any-questions-on-canal.md

@@ -0,0 +1,8 @@
+---
+name: Any Questions on Canal
+about: Ask whatever you want to know or confusion about Canal
+
+---
+
+## Question
+<!-- You can ask any question about this project -->

.github/ISSUE_TEMPLATE/bug-report-for-canal.md

@@ -0,0 +1,29 @@
+---
+name: Bug Report for Canal
+about: Create a report to help us improve
+
+---
+
+- [ ] I have searched the [issues](https://github.com/alibaba/canal/issues) of this repository and believe that this is not a duplicate.
+- [ ] I have checked the [FAQ](https://github.com/alibaba/canal/wiki/FAQ) of this repository and believe that this is not a duplicate.
+
+### environment
+
+* canal version
+* mysql version
+
+### Issue Description
+
+
+### Steps to reproduce
+
+### Expected behaviour
+
+### Actual behaviour
+
+
+If there is an exception, please attach the exception trace:
+
+```
+Just put your stack trace here!
+```

.github/ISSUE_TEMPLATE/feature-request-of-canal.md

@@ -0,0 +1,16 @@
+---
+name: Feature Request of Canal
+about: Suggest an idea for this project
+
+---
+
+## Why you need it?
+<!-- Is your feature request related to a problem? Please describe in details  -->
+
+
+## How it could be?
+<!--A clear and concise description of what you want to happen. For a computer fan,  you can explain more about input of the feature, and output of it.-->
+
+
+## Other related information
+<!-- Add any other context or screenshots about the feature request here.-->

.github/issue_template.md

@@ -1,13 +0,0 @@
-### environment
-
-* canal version
-* mysql version
-
-### Issue Description
-
-
-### Steps to reproduce
-
-### Expected behaviour
-
-### Actual behaviour

.gitignore

@@ -17,3 +17,4 @@ jtester.properties
 *.tar.gz
 *.rpm
 client-adapter/example/
+*.dat

.travis.yml

@@ -2,6 +2,8 @@ language: java
 sudo: false # faster builds
 
 jdk:
+  - openjdk11
+  - oraclejdk11
   - openjdk8
   - oraclejdk8
 
@@ -12,7 +14,7 @@ cache:
 install: true
 
 script:
-  - travis_wait 30 ./mvnw clean install -DskipTests=true -Dcheckstyle.skip=true -Drat.skip=true -Dmaven.javadoc.skip=true
+  - travis_wait 30 ./mvnw clean install -DskipTests=false -Dcheckstyle.skip=true -Dmaven.javadoc.skip=true
 
 after_success:
   - bash <(curl -s https://codecov.io/bash)

README.md

@@ -2,6 +2,7 @@
     <div class="iteye-blog-content-contain">
 
 [![Build Status](https://travis-ci.com/alibaba/canal.svg?branch=master)](https://travis-ci.com/alibaba/canal)
+[![codecov](https://codecov.io/gh/alibaba/canal/branch/master/graph/badge.svg)](https://codecov.io/gh/alibaba/canal)
 ![maven](https://img.shields.io/maven-central/v/com.alibaba.otter/canal.svg)
 ![license](https://img.shields.io/github/license/alibaba/canal.svg)
 [![Average time to resolve an issue](http://isitmaintained.com/badge/resolution/alibaba/canal.svg)](http://isitmaintained.com/project/alibaba/canal "Average time to resolve an issue")

client-adapter/common/pom.xml

@@ -1,10 +1,9 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.3-SNAPSHOT</version>
+        <version>1.1.4-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.alibaba.otter</groupId>
@@ -15,7 +14,7 @@
         <dependency>
             <groupId>com.alibaba.otter</groupId>
             <artifactId>canal.protocol</artifactId>
-            <version>1.1.3-SNAPSHOT</version>
+            <version>1.1.4-SNAPSHOT</version>
         </dependency>
         <dependency>
             <groupId>joda-time</groupId>

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

@@ -0,0 +1,121 @@
+package com.alibaba.otter.canal.client.adapter.support;
+
+import com.alibaba.druid.pool.DruidDataSource;
+import com.google.common.base.Joiner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.sql.DataSource;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicLong;
+
+public abstract class AbstractEtlService {
+
+    protected Logger      logger = LoggerFactory.getLogger(this.getClass());
+
+    private String        type;
+    private AdapterConfig config;
+    private final long CNT_PER_TASK = 10000L;
+
+    public AbstractEtlService(String type, AdapterConfig config){
+        this.type = type;
+        this.config = config;
+    }
+
+    protected EtlResult importData(String sql, List<String> params) {
+        EtlResult etlResult = new EtlResult();
+        AtomicLong impCount = new AtomicLong();
+        List<String> errMsg = new ArrayList<>();
+        if (config == null) {
+            logger.warn("{} mapping config is null, etl go end ", type);
+            etlResult.setErrorMessage(type + "mapping config is null, etl go end ");
+            return etlResult;
+        }
+
+        long start = System.currentTimeMillis();
+        try {
+            DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
+
+            List<Object> values = new ArrayList<>();
+            // 拼接条件
+            if (config.getMapping().getEtlCondition() != null && params != null) {
+                String etlCondition = config.getMapping().getEtlCondition();
+                int size = params.size();
+                for (String param : params) {
+                    etlCondition = etlCondition.replace("{}", "?");
+                    values.add(param);
+                }
+
+                sql += " " + etlCondition;
+            }
+
+            if (logger.isDebugEnabled()) {
+                logger.debug("etl sql : {}", sql);
+            }
+
+            // 获取总数
+            String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT ";
+            long cnt = (Long) Util.sqlRS(dataSource, countSql, values, rs -> {
+                Long count = null;
+                try {
+                    if (rs.next()) {
+                        count = ((Number) rs.getObject(1)).longValue();
+                    }
+                } catch (Exception e) {
+                    logger.error(e.getMessage(), e);
+                }
+                return count == null ? 0L : count;
+            });
+
+            // 当大于1万条记录时开启多线程
+            if (cnt >= 10000) {
+                int threadCount = Runtime.getRuntime().availableProcessors();
+
+                long offset;
+                long size = CNT_PER_TASK;
+                long workerCnt = cnt / size + (cnt % size == 0 ? 0 : 1);
+
+                if (logger.isDebugEnabled()) {
+                    logger.debug("workerCnt {} for cnt {} threadCount {}", workerCnt, cnt, threadCount);
+                }
+
+                ExecutorService executor = Util.newFixedThreadPool(threadCount, 5000L);
+                List<Future<Boolean>> futures = new ArrayList<>();
+                for (long i = 0; i < workerCnt; i++) {
+                    offset = size * i;
+                    String sqlFinal = sql + " LIMIT " + offset + "," + size;
+                    Future<Boolean> future = executor.submit(
+                        () -> executeSqlImport(dataSource, sqlFinal, values, config.getMapping(), impCount, errMsg));
+                    futures.add(future);
+                }
+
+                for (Future<Boolean> future : futures) {
+                    future.get();
+                }
+                executor.shutdown();
+            } else {
+                executeSqlImport(dataSource, sql, values, config.getMapping(), impCount, errMsg);
+            }
+
+            logger.info("数据全量导入完成, 一共导入 {} 条数据, 耗时: {}", impCount.get(), System.currentTimeMillis() - start);
+            etlResult.setResultMessage("导入" + type + " 数据：" + impCount.get() + " 条");
+        } catch (Exception e) {
+            logger.error(e.getMessage(), e);
+            errMsg.add(type + " 数据导入异常 =>" + e.getMessage());
+        }
+        if (errMsg.isEmpty()) {
+            etlResult.setSucceeded(true);
+        } else {
+            etlResult.setErrorMessage(Joiner.on("\n").join(errMsg));
+        }
+        return etlResult;
+    }
+
+    protected abstract boolean executeSqlImport(DataSource ds, String sql, List<Object> values,
+                                                AdapterConfig.AdapterMapping mapping, AtomicLong impCount,
+                                                List<String> errMsg);
+
+}

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

@@ -0,0 +1,11 @@
+package com.alibaba.otter.canal.client.adapter.support;
+
+public interface AdapterConfig {
+    String getDataSourceKey();
+
+    AdapterMapping getMapping();
+
+    interface AdapterMapping {
+        String getEtlCondition();
+    }
+}

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

@@ -33,7 +33,14 @@ public class CanalClientConfig {
     // aliyun ak/sk
     private String             accessKey;
     private String             secretKey;
-
+    // 是否启用消息轨迹
+    private boolean            enableMessageTrace;
+    // 在使用阿里云商业化mq服务时，如果想使用云上消息轨迹功能，请设置此配置为true
+    private String             accessChannel;
+    // 用于使用开源RocketMQ时，设置自定义的消息轨迹topic
+    private String             customizedTraceTopic;
+    // 开源RocketMQ命名空间
+    private String             namespace;
     // canal adapters 配置
     private List<CanalAdapter> canalAdapters;
 
@@ -133,6 +140,38 @@ public class CanalClientConfig {
         this.canalAdapters = canalAdapters;
     }
 
+    public boolean isEnableMessageTrace() {
+        return enableMessageTrace;
+    }
+
+    public void setEnableMessageTrace(boolean enableMessageTrace) {
+        this.enableMessageTrace = enableMessageTrace;
+    }
+
+    public String getAccessChannel() {
+        return accessChannel;
+    }
+
+    public void setAccessChannel(String accessChannel) {
+        this.accessChannel = accessChannel;
+    }
+
+    public String getCustomizedTraceTopic() {
+        return customizedTraceTopic;
+    }
+
+    public void setCustomizedTraceTopic(String customizedTraceTopic) {
+        this.customizedTraceTopic = customizedTraceTopic;
+    }
+
+    public String getNamespace() {
+        return namespace;
+    }
+
+    public void setNamespace(String namespace) {
+        this.namespace = namespace;
+    }
+
     public static class CanalAdapter {
 
         private String      instance; // 实例名

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

@@ -76,11 +76,16 @@ public class MessageUtil {
                                 dml.getPkNames().add(column.getName());
                             }
                         }
+                        if (column.getIsNull()) {
+                            row.put(column.getName(), null);
+                        } else {
                             row.put(column.getName(),
-                            JdbcTypeUtil.typeConvert(dml.getTable(),column.getName(),
+                                JdbcTypeUtil.typeConvert(dml.getTable(),
+                                    column.getName(),
                                     column.getValue(),
                                     column.getSqlType(),
                                     column.getMysqlType()));
+                        }
                         // 获取update为true的字段
                         if (column.getUpdated()) {
                             updateSet.add(column.getName());
@@ -94,13 +99,18 @@ public class MessageUtil {
                         Map<String, Object> rowOld = new LinkedHashMap<>();
                         for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
                             if (updateSet.contains(column.getName())) {
+                                if (column.getIsNull()) {
+                                    rowOld.put(column.getName(), null);
+                                } else {
                                     rowOld.put(column.getName(),
-                                    JdbcTypeUtil.typeConvert(dml.getTable(),column.getName(),
+                                        JdbcTypeUtil.typeConvert(dml.getTable(),
+                                            column.getName(),
                                             column.getValue(),
                                             column.getSqlType(),
                                             column.getMysqlType()));
                                 }
                             }
+                        }
                         // update操作将记录修改前的值
                         if (!rowOld.isEmpty()) {
                             old.add(rowOld);
@@ -162,8 +172,8 @@ public class MessageUtil {
         return dml;
     }
 
-    private static List<Map<String, Object>> changeRows(String table, List<Map<String, String>> rows, Map<String, Integer> sqlTypes,
+    private static List<Map<String, Object>> changeRows(String table, List<Map<String, String>> rows,
-                                                        Map<String, String> mysqlTypes) {
+                                                        Map<String, Integer> sqlTypes, Map<String, String> mysqlTypes) {
         List<Map<String, Object>> result = new ArrayList<>();
         for (Map<String, String> row : rows) {
             Map<String, Object> resultRow = new LinkedHashMap<>();

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

@@ -2,13 +2,11 @@ package com.alibaba.otter.canal.client.adapter.support;
 
 import java.io.File;
 import java.net.URL;
-import java.sql.Connection;
+import java.sql.*;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Statement;
 import java.time.*;
 import java.time.format.DateTimeFormatter;
 import java.util.Date;
+import java.util.List;
 import java.util.TimeZone;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -37,9 +35,9 @@ public class Util {
      */
     public static Object sqlRS(DataSource ds, String sql, Function<ResultSet, Object> fun) {
         try (Connection conn = ds.getConnection();
-                Statement stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);) {
+                Statement stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)) {
             stmt.setFetchSize(Integer.MIN_VALUE);
-            try (ResultSet rs = stmt.executeQuery(sql);) {
+            try (ResultSet rs = stmt.executeQuery(sql)) {
                 return fun.apply(rs);
             }
         } catch (Exception e) {
@@ -48,6 +46,26 @@ public class Util {
         }
     }
 
+    public static Object sqlRS(DataSource ds, String sql, List<Object> values, Function<ResultSet, Object> fun) {
+        try (Connection conn = ds.getConnection()) {
+            try (PreparedStatement pstmt = conn
+                .prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)) {
+                pstmt.setFetchSize(Integer.MIN_VALUE);
+                if (values != null) {
+                    for (int i = 0; i < values.size(); i++) {
+                        pstmt.setObject(i + 1, values.get(i));
+                    }
+                }
+                try (ResultSet rs = pstmt.executeQuery()) {
+                    return fun.apply(rs);
+                }
+            }
+        } catch (Exception e) {
+            logger.error("sqlRs has error, sql: {} ", sql);
+            throw new RuntimeException(e);
+        }
+    }
+
     /**
      * sql执行获取resultSet
      *

client-adapter/elasticsearch/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.3-SNAPSHOT</version>
+        <version>1.1.4-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.alibaba.otter</groupId>
@@ -21,7 +21,7 @@
         <dependency>
             <groupId>com.alibaba.fastsql</groupId>
             <artifactId>fastsql</artifactId>
-            <version>2.0.0_preview_644</version>
+            <version>2.0.0_preview_855</version>
         </dependency>
         <dependency>
             <groupId>org.elasticsearch</groupId>

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

@@ -30,7 +30,6 @@ import com.alibaba.otter.canal.client.adapter.es.service.ESEtlService;
 import com.alibaba.otter.canal.client.adapter.es.service.ESSyncService;
 import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate;
 import com.alibaba.otter.canal.client.adapter.support.*;
-import org.slf4j.MDC;
 
 /**
  * ES外部适配器
@@ -71,7 +70,6 @@ public class ESAdapter implements OuterAdapter {
     @Override
     public void init(OuterAdapterConfig configuration, Properties envProperties) {
         try {
-            MDC.put("adapter", "es");
             this.envProperties = envProperties;
             Map<String, ESSyncConfig> esSyncConfigTmp = ESSyncConfigLoader.load(envProperties);
             // 过滤不匹配的key的配置
@@ -141,6 +139,7 @@ public class ESAdapter implements OuterAdapter {
         }
     }
 
+    @Override
     public void sync(List<Dml> dmls) {
         if (dmls == null || dmls.isEmpty()) {
             return;
@@ -151,6 +150,7 @@ public class ESAdapter implements OuterAdapter {
             }
         }
         esSyncService.commit(); // 批次统一提交
+
     }
 
     private void sync(Dml dml) {
@@ -188,7 +188,6 @@ public class ESAdapter implements OuterAdapter {
         } else {
             StringBuilder resultMsg = new StringBuilder();
             boolean resSuccess = true;
-            // ds不为空说明传入的是datasourceKey
             for (ESSyncConfig configTmp : esSyncConfig.values()) {
                 // 取所有的destination为task的配置
                 if (configTmp.getDestination().equals(task)) {
@@ -241,7 +240,6 @@ public class ESAdapter implements OuterAdapter {
         if (transportClient != null) {
             transportClient.close();
         }
-        MDC.remove("adapter");
     }
 
     @Override

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

@@ -1,5 +1,7 @@
 package com.alibaba.otter.canal.client.adapter.es.config;
 
+import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
+
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -11,7 +13,7 @@ import java.util.Map;
  * @author rewerma 2018-11-01
  * @version 1.0.0
  */
-public class ESSyncConfig {
+public class ESSyncConfig implements AdapterConfig {
 
     private String    dataSourceKey;   // 数据源key
 
@@ -78,7 +80,11 @@ public class ESSyncConfig {
         this.esMapping = esMapping;
     }
 
-    public static class ESMapping {
+    public ESMapping getMapping(){
+        return esMapping;
+    }
+
+    public static class ESMapping implements AdapterMapping{
 
         private String                       _index;
         private String                       _type;

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

@@ -10,7 +10,11 @@ import java.util.stream.Collectors;
 import com.alibaba.fastsql.sql.SQLUtils;
 import com.alibaba.fastsql.sql.ast.SQLExpr;
 import com.alibaba.fastsql.sql.ast.expr.*;
-import com.alibaba.fastsql.sql.ast.statement.*;
+import com.alibaba.fastsql.sql.ast.statement.SQLExprTableSource;
+import com.alibaba.fastsql.sql.ast.statement.SQLJoinTableSource;
+import com.alibaba.fastsql.sql.ast.statement.SQLSelectStatement;
+import com.alibaba.fastsql.sql.ast.statement.SQLSubqueryTableSource;
+import com.alibaba.fastsql.sql.ast.statement.SQLTableSource;
 import com.alibaba.fastsql.sql.dialect.mysql.ast.statement.MySqlSelectQueryBlock;
 import com.alibaba.fastsql.sql.dialect.mysql.parser.MySqlStatementParser;
 import com.alibaba.fastsql.sql.parser.ParserException;
@@ -122,6 +126,23 @@ public class SqlParser {
             SQLCaseExpr sqlCaseExpr = (SQLCaseExpr) expr;
             fieldItem.setMethod(true);
             sqlCaseExpr.getItems().forEach(item-> visitColumn(item.getConditionExpr(), fieldItem));
+        }else if(expr instanceof SQLCharExpr) {
+            SQLCharExpr sqlCharExpr = (SQLCharExpr) expr;
+            String owner = null;
+            String columnName = null;
+            if (sqlCharExpr.getParent() instanceof SQLCaseExpr.Item) {
+                owner = ((SQLPropertyExpr) ((SQLCaseExpr.Item) sqlCharExpr.getParent()).getValueExpr()).getOwnernName();
+                columnName = ((SQLPropertyExpr) ((SQLCaseExpr.Item) sqlCharExpr.getParent()).getValueExpr()).getName();
+            }
+            if (fieldItem.getFieldName() == null) {
+                fieldItem.setFieldName(columnName);
+                fieldItem.setExpr(sqlCharExpr.toString());
+            }
+            ColumnItem columnItem = new ColumnItem();
+            columnItem.setColumnName(columnName);
+            columnItem.setOwner(owner);
+            fieldItem.getOwners().add(owner);
+            fieldItem.addColumn(columnItem);
         }
     }
 

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

@@ -1,12 +1,11 @@
 package com.alibaba.otter.canal.client.adapter.es.service;
 
 import java.sql.SQLException;
-import java.util.*;
+import java.util.HashMap;
-import java.util.concurrent.ExecutorService;
+import java.util.LinkedHashMap;
-import java.util.concurrent.TimeUnit;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
 
 import javax.sql.DataSource;
 
@@ -21,18 +20,15 @@ import org.elasticsearch.client.transport.TransportClient;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.search.SearchHit;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
-import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
 import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping;
 import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.FieldItem;
 import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate;
-import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
+import com.alibaba.otter.canal.client.adapter.support.AbstractEtlService;
+import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
 import com.alibaba.otter.canal.client.adapter.support.EtlResult;
 import com.alibaba.otter.canal.client.adapter.support.Util;
-import com.google.common.base.Joiner;
 
 /**
  * ES ETL Service
@@ -40,116 +36,24 @@ import com.google.common.base.Joiner;
  * @author rewerma 2018-11-01
  * @version 1.0.0
  */
-public class ESEtlService {
+public class ESEtlService extends AbstractEtlService {
-
-    private static Logger   logger = LoggerFactory.getLogger(ESEtlService.class);
 
     private TransportClient transportClient;
     private ESTemplate      esTemplate;
     private ESSyncConfig    config;
 
     public ESEtlService(TransportClient transportClient, ESSyncConfig config){
+        super("ES", config);
         this.transportClient = transportClient;
         this.esTemplate = new ESTemplate(transportClient);
         this.config = config;
     }
 
     public EtlResult importData(List<String> params) {
-        EtlResult etlResult = new EtlResult();
-        AtomicLong impCount = new AtomicLong();
-        List<String> errMsg = new ArrayList<>();
-        String esIndex = "";
-        if (config == null) {
-            logger.warn("esSycnCofnig is null, etl go end !");
-            etlResult.setErrorMessage("esSycnCofnig is null, etl go end !");
-            return etlResult;
-        }
-
         ESMapping mapping = config.getEsMapping();
-
+        logger.info("start etl to import data to index: {}", mapping.get_index());
-        esIndex = mapping.get_index();
-        DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
-        Pattern pattern = Pattern.compile(".*:(.*)://.*/(.*)\\?.*$");
-        Matcher matcher = pattern.matcher(dataSource.getUrl());
-        if (!matcher.find()) {
-            throw new RuntimeException("Not found the schema of jdbc-url: " + config.getDataSourceKey());
-        }
-        String schema = matcher.group(2);
-
-        logger.info("etl from db: {},  to es index: {}", schema, esIndex);
-        long start = System.currentTimeMillis();
-        try {
         String sql = mapping.getSql();
-
+        return importData(sql, params);
-            // 拼接条件
-            if (mapping.getEtlCondition() != null && params != null) {
-                String etlCondition = mapping.getEtlCondition();
-                int size = params.size();
-                for (int i = 0; i < size; i++) {
-                    etlCondition = etlCondition.replace("{" + i + "}", params.get(i));
-                }
-
-                sql += " " + etlCondition;
-            }
-
-            if (logger.isDebugEnabled()) {
-                logger.debug("etl sql : {}", mapping.getSql());
-            }
-
-            // 获取总数
-            String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT ";
-            long cnt = (Long) Util.sqlRS(dataSource, countSql, rs -> {
-                Long count = null;
-                try {
-                    if (rs.next()) {
-                        count = ((Number) rs.getObject(1)).longValue();
-                    }
-                } catch (Exception e) {
-                    logger.error(e.getMessage(), e);
-                }
-                return count == null ? 0L : count;
-            });
-
-            // 当大于1万条记录时开启多线程
-            if (cnt >= 10000) {
-                int threadCount = 3; // 从配置读取默认为3
-                long perThreadCnt = cnt / threadCount;
-                ExecutorService executor = Util.newFixedThreadPool(threadCount, 5000L);
-                for (int i = 0; i < threadCount; i++) {
-                    long offset = i * perThreadCnt;
-                    Long size = null;
-                    if (i != threadCount - 1) {
-                        size = perThreadCnt;
-                    }
-                    String sqlFinal;
-                    if (size != null) {
-                        sqlFinal = sql + " LIMIT " + offset + "," + size;
-                    } else {
-                        sqlFinal = sql + " LIMIT " + offset + "," + cnt;
-                    }
-                    executor.execute(() -> executeSqlImport(dataSource, sqlFinal, mapping, impCount, errMsg));
-                }
-
-                executor.shutdown();
-                while (!executor.awaitTermination(3, TimeUnit.SECONDS)) {
-                    // ignore
-                }
-            } else {
-                executeSqlImport(dataSource, sql, mapping, impCount, errMsg);
-            }
-
-            logger.info("数据全量导入完成,一共导入 {} 条数据, 耗时: {}", impCount.get(), System.currentTimeMillis() - start);
-            etlResult.setResultMessage("导入ES索引 " + esIndex + " 数据：" + impCount.get() + " 条");
-        } catch (Exception e) {
-            logger.error(e.getMessage(), e);
-            errMsg.add(esIndex + " etl failed! ==>" + e.getMessage());
-        }
-        if (errMsg.isEmpty()) {
-            etlResult.setSucceeded(true);
-        } else {
-            etlResult.setErrorMessage(Joiner.on("\n").join(errMsg));
-        }
-        return etlResult;
     }
 
     private void processFailBulkResponse(BulkResponse bulkResponse) {
@@ -167,10 +71,12 @@ public class ESEtlService {
         }
     }
 
-    private boolean executeSqlImport(DataSource ds, String sql, ESMapping mapping, AtomicLong impCount,
+    protected boolean executeSqlImport(DataSource ds, String sql, List<Object> values,
+                                       AdapterConfig.AdapterMapping adapterMapping, AtomicLong impCount,
                                        List<String> errMsg) {
         try {
-            Util.sqlRS(ds, sql, rs -> {
+            ESMapping mapping = (ESMapping) adapterMapping;
+            Util.sqlRS(ds, sql, values, rs -> {
                 int count = 0;
                 try {
                     BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();

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

@@ -1,13 +1,9 @@
 package com.alibaba.otter.canal.client.adapter.es.service;
 
-import java.util.Collection;
+import java.util.*;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
 
 import javax.sql.DataSource;
 
-import com.alibaba.otter.canal.client.adapter.support.Util;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -23,6 +19,7 @@ import com.alibaba.otter.canal.client.adapter.es.support.ESSyncUtil;
 import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate;
 import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 import com.alibaba.otter.canal.client.adapter.support.Dml;
+import com.alibaba.otter.canal.client.adapter.support.Util;
 
 /**
  * ES 同步 Service
@@ -436,7 +433,7 @@ public class ESSyncService {
         Object idVal = esTemplate.getESDataFromDmlData(mapping, data, esFieldData);
 
         if (logger.isTraceEnabled()) {
-            logger.trace("Single table insert ot es index, destination:{}, table: {}, index: {}, id: {}",
+            logger.trace("Single table insert to es index, destination:{}, table: {}, index: {}, id: {}",
                 config.getDestination(),
                 dml.getTable(),
                 mapping.get_index(),
@@ -459,7 +456,7 @@ public class ESSyncService {
         sql = ESSyncUtil.appendCondition(sql, condition);
         DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
         if (logger.isTraceEnabled()) {
-            logger.trace("Main table insert ot es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
+            logger.trace("Main table insert to es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
                 config.getDestination(),
                 dml.getTable(),
                 mapping.get_index(),
@@ -473,7 +470,7 @@ public class ESSyncService {
 
                     if (logger.isTraceEnabled()) {
                         logger.trace(
-                            "Main table insert ot es index by query sql, destination:{}, table: {}, index: {}, id: {}",
+                            "Main table insert to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
                             config.getDestination(),
                             dml.getTable(),
                             mapping.get_index(),
@@ -517,7 +514,7 @@ public class ESSyncService {
 
                     if (logger.isTraceEnabled()) {
                         logger.trace(
-                            "Main table delete ot es index by query sql, destination:{}, table: {}, index: {}, id: {}",
+                            "Main table delete to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
                             config.getDestination(),
                             dml.getTable(),
                             mapping.get_index(),
@@ -677,7 +674,17 @@ public class ESSyncService {
     private void wholeSqlOperation(ESSyncConfig config, Dml dml, Map<String, Object> data, Map<String, Object> old,
                                    TableItem tableItem) {
         ESMapping mapping = config.getEsMapping();
-        StringBuilder sql = new StringBuilder(mapping.getSql() + " WHERE ");
+        //防止最后出现groupby 导致sql解析异常
+        String[] sqlSplit = mapping.getSql().split("GROUP\\ BY(?!(.*)ON)");
+        String sqlNoWhere = sqlSplit[0];
+
+        String sqlGroupBy = "";
+
+        if(sqlSplit.length > 1){
+            sqlGroupBy =  "GROUP BY "+ sqlSplit[1];
+        }
+
+        StringBuilder sql = new StringBuilder(sqlNoWhere + " WHERE ");
 
         for (FieldItem fkFieldItem : tableItem.getRelationTableFields().keySet()) {
             String columnName = fkFieldItem.getColumn().getColumnName();
@@ -686,6 +693,8 @@ public class ESSyncService {
         }
         int len = sql.length();
         sql.delete(len - 5, len);
+        sql.append(sqlGroupBy);
+
         DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
         if (logger.isTraceEnabled()) {
             logger.trace("Join table update es index by query whole sql, destination:{}, table: {}, index: {}, sql: {}",
@@ -784,7 +793,7 @@ public class ESSyncService {
         Object idVal = esTemplate.getESDataFromDmlData(mapping, data, old, esFieldData);
 
         if (logger.isTraceEnabled()) {
-            logger.trace("Main table update ot es index, destination:{}, table: {}, index: {}, id: {}",
+            logger.trace("Main table update to es index, destination:{}, table: {}, index: {}, id: {}",
                 config.getDestination(),
                 dml.getTable(),
                 mapping.get_index(),
@@ -807,7 +816,7 @@ public class ESSyncService {
         sql = ESSyncUtil.appendCondition(sql, condition);
         DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
         if (logger.isTraceEnabled()) {
-            logger.trace("Main table update ot es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
+            logger.trace("Main table update to es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
                 config.getDestination(),
                 dml.getTable(),
                 mapping.get_index(),
@@ -821,7 +830,7 @@ public class ESSyncService {
 
                     if (logger.isTraceEnabled()) {
                         logger.trace(
-                            "Main table update ot es index by query sql, destination:{}, table: {}, index: {}, id: {}",
+                            "Main table update to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
                             config.getDestination(),
                             dml.getTable(),
                             mapping.get_index(),

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

@@ -5,7 +5,13 @@ import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
 import java.sql.Blob;
 import java.sql.SQLException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Set;
 
 import org.apache.commons.codec.binary.Base64;
 import org.joda.time.DateTime;
@@ -203,7 +209,7 @@ public class ESSyncUtil {
                     if (value.contains(";")) {
                         separator = ";";
                     } else if (value.contains("|")) {
-                        separator = "|";
+                        separator = "\\|";
                     } else if (value.contains("-")) {
                         separator = "-";
                     }

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

@@ -2,9 +2,7 @@ package com.alibaba.otter.canal.client.adapter.es.support;
 
 import java.sql.ResultSet;
 import java.sql.SQLException;
-import java.util.HashMap;
+import java.util.*;
-import java.util.LinkedHashMap;
-import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
@@ -60,6 +58,10 @@ public class ESTemplate {
         return bulkRequestBuilder;
     }
 
+    public void resetBulkRequestBuilder() {
+        this.bulkRequestBuilder = this.transportClient.prepareBulk();
+    }
+
     /**
      * 插入数据
      *
@@ -136,11 +138,15 @@ public class ESTemplate {
         // 查询sql批量更新
         DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
         StringBuilder sql = new StringBuilder("SELECT * FROM (" + mapping.getSql() + ") _v WHERE ");
-        paramsTmp.forEach(
+        List<Object> values = new ArrayList<>();
-            (fieldName, value) -> sql.append("_v.").append(fieldName).append("=").append(value).append(" AND "));
+        paramsTmp.forEach((fieldName, value) -> {
+            sql.append("_v.").append(fieldName).append("=? AND ");
+            values.add(value);
+        });
+        //TODO 直接外部包裹sql会导致全表扫描性能低, 待优化拼接内部where条件
         int len = sql.length();
         sql.delete(len - 4, len);
-        Integer syncCount = (Integer) Util.sqlRS(ds, sql.toString(), rs -> {
+        Integer syncCount = (Integer) Util.sqlRS(ds, sql.toString(), values, rs -> {
             int count = 0;
             try {
                 while (rs.next()) {
@@ -208,11 +214,12 @@ public class ESTemplate {
     }
 
     /**
-     * 如果大于批量数则提交批次
+     * 如果大于批量数则提交批次, 调用后es bulk请求后，numberOfActions不会清理，需要主动调用函数清0，否则不能起到批量请求的效果
      */
     private void commitBulk() {
         if (getBulk().numberOfActions() >= MAX_BATCH_SIZE) {
             commit();
+            resetBulkRequestBuilder();
         }
     }
 

client-adapter/elasticsearch/src/main/resources/es/biz_order.yml

@@ -17,5 +17,5 @@ esMapping:
         from biz_order t"
   skips:
     - customer_id
-  etlCondition: "where t.c_time>='{0}'"
+  etlCondition: "where t.c_time>={}"
   commitBatch: 3000

client-adapter/elasticsearch/src/main/resources/es/customer.yml

@@ -9,7 +9,7 @@ esMapping:
     customer_order:
       name: customer
   sql: "select t.id, t.name, t.email from customer t"
-  etlCondition: "where t.c_time>='{0}'"
+  etlCondition: "where t.c_time>={}"
   commitBatch: 3000
 
 

client-adapter/elasticsearch/src/main/resources/es/mytest_user.yml

@@ -12,5 +12,5 @@ esMapping:
         left join role b on b.id=a.role_id"
 #  objFields:
 #    _labels: array:;
-  etlCondition: "where a.c_time>='{0}'"
+  etlCondition: "where a.c_time>={}"
   commitBatch: 3000

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

@@ -4,12 +4,13 @@ import java.util.Map;
 
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.Ignore;
 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.DatasourceConfig;
-
+@Ignore
 public class ConfigLoadTest {
 
     @Before

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

@@ -18,8 +18,9 @@ import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.transport.client.PreBuiltTransportClient;
 import org.junit.After;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
-
+@Ignore
 public class ESTest {
 
     private TransportClient transportClient;

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

@@ -1,19 +1,24 @@
 package com.alibaba.otter.canal.client.adapter.es.test.sync;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
 
 import javax.sql.DataSource;
 
 import org.elasticsearch.action.get.GetResponse;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import com.alibaba.otter.canal.client.adapter.es.ESAdapter;
 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;
-
+@Ignore
 public class LabelSyncJoinSub2Test {
 
     private ESAdapter esAdapter;

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

@@ -1,19 +1,24 @@
 package com.alibaba.otter.canal.client.adapter.es.test.sync;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
 
 import javax.sql.DataSource;
 
 import org.elasticsearch.action.get.GetResponse;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import com.alibaba.otter.canal.client.adapter.es.ESAdapter;
 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;
-
+@Ignore
 public class LabelSyncJoinSubTest {
 
     private ESAdapter esAdapter;

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

@@ -1,19 +1,24 @@
 package com.alibaba.otter.canal.client.adapter.es.test.sync;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
 
 import javax.sql.DataSource;
 
 import org.elasticsearch.action.get.GetResponse;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import com.alibaba.otter.canal.client.adapter.es.ESAdapter;
 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;
-
+@Ignore
 public class RoleSyncJoinOne2Test {
 
     private ESAdapter esAdapter;

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

@@ -1,19 +1,24 @@
 package com.alibaba.otter.canal.client.adapter.es.test.sync;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
 
 import javax.sql.DataSource;
 
 import org.elasticsearch.action.get.GetResponse;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import com.alibaba.otter.canal.client.adapter.es.ESAdapter;
 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;
-
+@Ignore
 public class RoleSyncJoinOneTest {
 
     private ESAdapter esAdapter;

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

@@ -1,19 +1,24 @@
 package com.alibaba.otter.canal.client.adapter.es.test.sync;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
 
 import javax.sql.DataSource;
 
 import org.elasticsearch.action.get.GetResponse;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import com.alibaba.otter.canal.client.adapter.es.ESAdapter;
 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;
-
+@Ignore
 public class UserSyncJoinOneTest {
 
     private ESAdapter esAdapter;

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

@@ -1,16 +1,21 @@
 package com.alibaba.otter.canal.client.adapter.es.test.sync;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
 
 import org.elasticsearch.action.get.GetResponse;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import com.alibaba.otter.canal.client.adapter.es.ESAdapter;
 import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
 import com.alibaba.otter.canal.client.adapter.support.Dml;
-
+@Ignore
 public class UserSyncSingleTest {
 
     private ESAdapter esAdapter;

client-adapter/hbase/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.3-SNAPSHOT</version>
+        <version>1.1.4-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.alibaba.otter</groupId>
@@ -26,6 +26,10 @@
                     <groupId>org.slf4j</groupId>
                     <artifactId>slf4j-log4j12</artifactId>
                 </exclusion>
+                <exclusion>
+                    <groupId>jdk.tools</groupId>
+                    <artifactId>jdk.tools</artifactId>
+                </exclusion>
             </exclusions>
         </dependency>
         <dependency>

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

@@ -4,8 +4,6 @@ import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 
-import javax.sql.DataSource;
-
 import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HBaseConfiguration;
@@ -25,8 +23,10 @@ import com.alibaba.otter.canal.client.adapter.hbase.monitor.HbaseConfigMonitor;
 import com.alibaba.otter.canal.client.adapter.hbase.service.HbaseEtlService;
 import com.alibaba.otter.canal.client.adapter.hbase.service.HbaseSyncService;
 import com.alibaba.otter.canal.client.adapter.hbase.support.HbaseTemplate;
-import com.alibaba.otter.canal.client.adapter.support.*;
+import com.alibaba.otter.canal.client.adapter.support.Dml;
-import org.slf4j.MDC;
+import com.alibaba.otter.canal.client.adapter.support.EtlResult;
+import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
+import com.alibaba.otter.canal.client.adapter.support.SPI;
 
 /**
  * HBase外部适配器
@@ -60,7 +60,6 @@ public class HbaseAdapter implements OuterAdapter {
     @Override
     public void init(OuterAdapterConfig configuration, Properties envProperties) {
         try {
-            MDC.put("adapter", "hbase");
             this.envProperties = envProperties;
             Map<String, MappingConfig> hbaseMappingTmp = MappingConfigLoader.load(envProperties);
             // 过滤不匹配的key的配置
@@ -104,6 +103,7 @@ public class HbaseAdapter implements OuterAdapter {
         }
     }
 
+    @Override
     public void sync(List<Dml> dmls) {
         if (dmls == null || dmls.isEmpty()) {
             return;
@@ -113,7 +113,7 @@ public class HbaseAdapter implements OuterAdapter {
         }
     }
 
-    public void sync(Dml dml) {
+    private void sync(Dml dml) {
         if (dml == null) {
             return;
         }
@@ -148,27 +148,16 @@ public class HbaseAdapter implements OuterAdapter {
     public EtlResult etl(String task, List<String> params) {
         EtlResult etlResult = new EtlResult();
         MappingConfig config = hbaseMapping.get(task);
+        HbaseEtlService hbaseEtlService = new HbaseEtlService(hbaseTemplate, config);
         if (config != null) {
-            DataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
+            return hbaseEtlService.importData(params);
-            if (dataSource != null) {
-                return HbaseEtlService.importData(dataSource, hbaseTemplate, config, params);
-            } else {
-                etlResult.setSucceeded(false);
-                etlResult.setErrorMessage("DataSource not found");
-                return etlResult;
-            }
         } else {
             StringBuilder resultMsg = new StringBuilder();
             boolean resSucc = true;
-            // ds不为空说明传入的是datasourceKey
             for (MappingConfig configTmp : hbaseMapping.values()) {
                 // 取所有的destination为task的配置
                 if (configTmp.getDestination().equals(task)) {
-                    DataSource dataSource = DatasourceConfig.DATA_SOURCES.get(configTmp.getDataSourceKey());
+                    EtlResult etlRes = hbaseEtlService.importData(params);
-                    if (dataSource == null) {
-                        continue;
-                    }
-                    EtlResult etlRes = HbaseEtlService.importData(dataSource, hbaseTemplate, configTmp, params);
                     if (!etlRes.getSucceeded()) {
                         resSucc = false;
                         resultMsg.append(etlRes.getErrorMessage()).append("\n");
@@ -224,7 +213,6 @@ public class HbaseAdapter implements OuterAdapter {
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
-        MDC.remove("adapter");
     }
 
     @Override

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

@@ -1,5 +1,7 @@
 package com.alibaba.otter.canal.client.adapter.hbase.config;
 
+import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
+
 import java.util.*;
 
 /**
@@ -8,7 +10,7 @@ import java.util.*;
  * @author rewerma 2018-8-21 下午06:45:49
  * @version 1.0.0
  */
-public class MappingConfig {
+public class MappingConfig implements AdapterConfig {
 
     private String       dataSourceKey;   // 数据源key
 
@@ -60,6 +62,10 @@ public class MappingConfig {
         this.hbaseMapping = hbaseMapping;
     }
 
+    public AdapterMapping getMapping() {
+        return hbaseMapping;
+    }
+
     public void validate() {
         if (hbaseMapping.database == null || hbaseMapping.database.isEmpty()) {
             throw new NullPointerException("hbaseMapping.database");
@@ -179,7 +185,7 @@ public class MappingConfig {
         }
     }
 
-    public static class HbaseMapping {
+    public static class HbaseMapping implements AdapterMapping {
 
         private Mode                    mode               = Mode.STRING;           // hbase默认转换格式
         private String                  database;                                   // 数据库名或schema名

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

@@ -1,31 +1,23 @@
 package com.alibaba.otter.canal.client.adapter.hbase.service;
 
-import java.sql.ResultSetMetaData;
 import java.sql.Timestamp;
 import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
-import java.util.concurrent.*;
+import java.util.concurrent.ExecutorService;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicLong;
 
 import javax.sql.DataSource;
 
+import com.alibaba.otter.canal.client.adapter.support.*;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig;
-import com.alibaba.otter.canal.client.adapter.hbase.support.HRow;
+import com.alibaba.otter.canal.client.adapter.hbase.support.*;
-import com.alibaba.otter.canal.client.adapter.hbase.support.HbaseTemplate;
-import com.alibaba.otter.canal.client.adapter.hbase.support.PhType;
-import com.alibaba.otter.canal.client.adapter.hbase.support.PhTypeUtil;
-import com.alibaba.otter.canal.client.adapter.hbase.support.Type;
-import com.alibaba.otter.canal.client.adapter.hbase.support.TypeUtil;
-import com.alibaba.otter.canal.client.adapter.support.EtlResult;
-import com.alibaba.otter.canal.client.adapter.support.JdbcTypeUtil;
-import com.alibaba.otter.canal.client.adapter.support.Util;
 import com.google.common.base.Joiner;
 
 /**
@@ -34,17 +26,21 @@ import com.google.common.base.Joiner;
  * @author rewerma @ 2018-10-20
  * @version 1.0.0
  */
-public class HbaseEtlService {
+public class HbaseEtlService extends AbstractEtlService {
 
-    private static Logger logger = LoggerFactory.getLogger(HbaseEtlService.class);
+    private HbaseTemplate hbaseTemplate;
+    private MappingConfig config;
+
+    public HbaseEtlService(HbaseTemplate hbaseTemplate, MappingConfig config){
+        super("HBase", config);
+        this.hbaseTemplate = hbaseTemplate;
+        this.config = config;
+    }
 
     /**
      * 建表
-     *
-     * @param hbaseTemplate
-     * @param config
      */
-    public static void createTable(HbaseTemplate hbaseTemplate, MappingConfig config) {
+    private void createTable() {
         try {
             // 判断hbase表是否存在，不存在则建表
             MappingConfig.HbaseMapping hbaseMapping = config.getHbaseMapping();
@@ -60,29 +56,14 @@ public class HbaseEtlService {
     /**
      * 导入数据
      *
-     * @param ds 数据源
-     * @param hbaseTemplate hbaseTemplate
-     * @param config 配置
      * @param params 筛选条件
      * @return 导入结果
      */
-    public static EtlResult importData(DataSource ds, HbaseTemplate hbaseTemplate, MappingConfig config,
+    public EtlResult importData(List<String> params) {
-                                       List<String> params) {
         EtlResult etlResult = new EtlResult();
-        AtomicLong successCount = new AtomicLong();
         List<String> errMsg = new ArrayList<>();
-        String hbaseTable = "";
         try {
-            if (config == null) {
-                logger.error("Config is null!");
-                etlResult.setSucceeded(false);
-                etlResult.setErrorMessage("Config is null!");
-                return etlResult;
-            }
             MappingConfig.HbaseMapping hbaseMapping = config.getHbaseMapping();
-            hbaseTable = hbaseMapping.getHbaseTable();
-
-            long start = System.currentTimeMillis();
 
             if (params != null && params.size() == 1 && "rebuild".equalsIgnoreCase(params.get(0))) {
                 logger.info(hbaseMapping.getHbaseTable() + " rebuild is starting!");
@@ -95,119 +76,28 @@ public class HbaseEtlService {
             } else {
                 logger.info(hbaseMapping.getHbaseTable() + " etl is starting!");
             }
-            createTable(hbaseTemplate, config);
+            createTable();
 
             // 拼接sql
             String sql = "SELECT * FROM " + config.getHbaseMapping().getDatabase() + "." + hbaseMapping.getTable();
 
-            // 拼接条件
+            return super.importData(sql, params);
-            if (params != null && params.size() == 1 && hbaseMapping.getEtlCondition() == null) {
-                AtomicBoolean stExists = new AtomicBoolean(false);
-                // 验证是否有SYS_TIME字段
-                Util.sqlRS(ds, sql, rs -> {
-                    try {
-                        ResultSetMetaData rsmd = rs.getMetaData();
-                        int cnt = rsmd.getColumnCount();
-                        for (int i = 1; i <= cnt; i++) {
-                            String columnName = rsmd.getColumnName(i);
-                            if ("SYS_TIME".equalsIgnoreCase(columnName)) {
-                                stExists.set(true);
-                                break;
-                            }
-                        }
-                    } catch (Exception e) {
-                        // ignore
-                    }
-                    return null;
-                });
-                if (stExists.get()) {
-                    sql += " WHERE SYS_TIME >= '" + params.get(0) + "' ";
-                }
-            } else if (hbaseMapping.getEtlCondition() != null && params != null) {
-                String etlCondition = hbaseMapping.getEtlCondition();
-                int size = params.size();
-                for (int i = 0; i < size; i++) {
-                    etlCondition = etlCondition.replace("{" + i + "}", params.get(i));
-                }
-
-                sql += " " + etlCondition;
-            }
-
-            // 获取总数
-            String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT ";
-            long cnt = (Long) Util.sqlRS(ds, countSql, rs -> {
-                Long count = null;
-                try {
-                    if (rs.next()) {
-                        count = ((Number) rs.getObject(1)).longValue();
-                    }
         } catch (Exception e) {
             logger.error(e.getMessage(), e);
+            errMsg.add("HBase etl error ==>" + e.getMessage());
         }
-                return count == null ? 0 : count;
-            });
-
-            // 当大于1万条记录时开启多线程
-            if (cnt >= 10000) {
-                int threadCount = 3;
-                long perThreadCnt = cnt / threadCount;
-                ExecutorService executor = Util.newFixedThreadPool(threadCount, 5000L);
-                for (int i = 0; i < threadCount; i++) {
-                    long offset = i * perThreadCnt;
-                    Long size = null;
-                    if (i != threadCount - 1) {
-                        size = perThreadCnt;
-                    }
-                    String sqlFinal;
-                    if (size != null) {
-                        sqlFinal = sql + " LIMIT " + offset + "," + size;
-                    } else {
-                        sqlFinal = sql + " LIMIT " + offset + "," + cnt;
-                    }
-                    executor.submit(
-                        () -> executeSqlImport(ds, sqlFinal, hbaseMapping, hbaseTemplate, successCount, errMsg));
-                }
-
-                executor.shutdown();
-                while (!executor.awaitTermination(3, TimeUnit.SECONDS)) {
-                    // ignore
-                }
-            } else {
-                executeSqlImport(ds, sql, hbaseMapping, hbaseTemplate, successCount, errMsg);
-            }
-
-            logger.info(hbaseMapping.getHbaseTable() + " etl completed in: "
-                        + (System.currentTimeMillis() - start) / 1000 + "s!");
-
-            etlResult.setResultMessage("导入HBase表 " + hbaseMapping.getHbaseTable() + " 数据：" + successCount.get() + " 条");
-        } catch (Exception e) {
-            logger.error(e.getMessage(), e);
-            errMsg.add(hbaseTable + " etl failed! ==>" + e.getMessage());
-        }
-
-        if (errMsg.isEmpty()) {
-            etlResult.setSucceeded(true);
-        } else {
         etlResult.setErrorMessage(Joiner.on("\n").join(errMsg));
-        }
         return etlResult;
     }
 
     /**
      * 执行导入
-     *
-     * @param ds
-     * @param sql
-     * @param hbaseMapping
-     * @param hbaseTemplate
-     * @param successCount
-     * @param errMsg
-     * @return
      */
-    private static boolean executeSqlImport(DataSource ds, String sql, MappingConfig.HbaseMapping hbaseMapping,
+    protected boolean executeSqlImport(DataSource ds, String sql, List<Object> values,
-                                            HbaseTemplate hbaseTemplate, AtomicLong successCount, List<String> errMsg) {
+                                       AdapterConfig.AdapterMapping mapping, AtomicLong impCount, List<String> errMsg) {
+        MappingConfig.HbaseMapping hbaseMapping = (MappingConfig.HbaseMapping) mapping;
         try {
-            Util.sqlRS(ds, sql, rs -> {
+            Util.sqlRS(ds, sql, values, rs -> {
                 int i = 1;
 
                 try {
@@ -353,9 +243,9 @@ public class HbaseEtlService {
                             complete = true;
                         }
                         i++;
-                        successCount.incrementAndGet();
+                        impCount.incrementAndGet();
                         if (logger.isDebugEnabled()) {
-                            logger.debug("successful import count:" + successCount.get());
+                            logger.debug("successful import count:" + impCount.get());
                         }
                     }
 

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

@@ -130,8 +130,10 @@ public class PhTypeUtil {
             } else if (b[0] == 0) {
                 v = false;
             }
-        } else if (phType == PhType.TIME || phType == PhType.DATE) {
+        } else if (phType == PhType.DATE) {
-            v = new Date(decodeLong(b, 0));
+            v = new java.sql.Date(decodeLong(b, 0));
+        } else if (phType == PhType.TIME) {
+            v = new java.sql.Time(decodeLong(b, 0));
         } else if (phType == PhType.TIMESTAMP) {
             long millisDeserialized = decodeLong(b, 0);
             Timestamp ts = new Timestamp(millisDeserialized);

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

@@ -8,6 +8,7 @@ hbaseMapping:
   hbaseTable: MYTEST.PERSON2   # HBase表名
   family: CF  # 默认统一Family名称
   uppercaseQualifier: true  # 字段名转大写, 默认为true
+  etlCondition: "where c_time>={}"
   commitBatch: 3000 # 批量提交的大小
   #rowKey: id,type  # 复合字段rowKey不能和columns中的rowKey重复
   columns:

client-adapter/hbase/src/test/java/com/alibaba/otter/canal/client/adapter/hbase/test/HBaseConnectionTest.java

@@ -0,0 +1,19 @@
+package com.alibaba.otter.canal.client.adapter.hbase.test;
+
+import com.alibaba.otter.canal.client.adapter.hbase.support.HbaseTemplate;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.junit.Test;
+
+public class HBaseConnectionTest {
+
+    @Test
+    public void test01() {
+        Configuration hbaseConfig = HBaseConfiguration.create();
+        hbaseConfig.set("hbase.zookeeper.quorum", "127.0.0.1");
+        hbaseConfig.set("hbase.zookeeper.property.clientPort", "2181");
+        hbaseConfig.set("zookeeper.znode.parent", "/hbase");
+        HbaseTemplate hbaseTemplate = new HbaseTemplate(hbaseConfig);
+        System.out.println(hbaseTemplate.tableExists("ttt"));
+    }
+}

client-adapter/launcher/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.3-SNAPSHOT</version>
+        <version>1.1.4-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.alibaba.otter</groupId>
@@ -32,7 +32,7 @@
         <dependency>
             <groupId>com.alibaba.otter</groupId>
             <artifactId>canal.client</artifactId>
-            <version>1.1.3-SNAPSHOT</version>
+            <version>1.1.4-SNAPSHOT</version>
         </dependency>
         <dependency>
             <groupId>org.yaml</groupId>
@@ -62,7 +62,7 @@
         <dependency>
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-client</artifactId>
-            <version>4.3.0</version>
+	    <version>4.5.1</version>
         </dependency>
         <!-- 单独引入kafka依赖 -->
         <dependency>

client-adapter/launcher/src/main/bin/startup.sh

@@ -76,7 +76,7 @@ echo "cd to $bin_abs_path for workaround relative path"
 cd $bin_abs_path
 
 echo CLASSPATH :$CLASSPATH
-$JAVA $JAVA_OPTS $JAVA_DEBUG_OPT $ADAPTER_OPTS -classpath .:$CLASSPATH com.alibaba.otter.canal.adapter.launcher.CanalAdapterApplication 1>>$base/logs/adapter.log 2>&1 &
+$JAVA $JAVA_OPTS $JAVA_DEBUG_OPT $ADAPTER_OPTS -classpath .:$CLASSPATH com.alibaba.otter.canal.adapter.launcher.CanalAdapterApplication 1>>/dev/null 2>&1 &
 echo $! > $base/bin/adapter.pid
 
 echo "cd to $current_path for continue"

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

@@ -2,7 +2,11 @@ package com.alibaba.otter.canal.adapter.launcher.loader;
 
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -126,7 +130,11 @@ public class CanalAdapterLoader {
                         canalOuterAdapterGroups,
                         canalClientConfig.getAccessKey(),
                         canalClientConfig.getSecretKey(),
-                        canalClientConfig.getFlatMessage());
+                        canalClientConfig.getFlatMessage(),
+                        canalClientConfig.isEnableMessageTrace(),
+                        canalClientConfig.getCustomizedTraceTopic(),
+                        canalClientConfig.getAccessChannel(),
+                        canalClientConfig.getNamespace());
                     canalMQWorker.put(canalAdapter.getInstance() + "-rocketmq-" + group.getGroupId(), rocketMQWorker);
                     rocketMQWorker.start();
 

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

@@ -40,6 +40,30 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
         logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId);
     }
 
+    public CanalAdapterRocketMQWorker(CanalClientConfig canalClientConfig, String nameServers, String topic,
+        String groupId, List<List<OuterAdapter>> canalOuterAdapters, String accessKey,
+        String secretKey, boolean flatMessage, boolean enableMessageTrace,
+        String customizedTraceTopic, String accessChannel, String namespace) {
+        super(canalOuterAdapters);
+        this.canalClientConfig = canalClientConfig;
+        this.topic = topic;
+        this.flatMessage = flatMessage;
+        super.canalDestination = topic;
+        super.groupId = groupId;
+        this.connector = new RocketMQCanalConnector(nameServers,
+            topic,
+            groupId,
+            accessKey,
+            secretKey,
+            canalClientConfig.getBatchSize(),
+            flatMessage,
+            enableMessageTrace,
+            customizedTraceTopic,
+            accessChannel,
+            namespace);
+        logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId);
+    }
+
     @Override
     protected void process() {
         while (!running) {

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

@@ -1,13 +1,23 @@
 package com.alibaba.otter.canal.adapter.launcher.rest;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.Resource;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.web.bind.annotation.*;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.PutMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
 
 import com.alibaba.otter.canal.adapter.launcher.common.EtlLock;
 import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch;

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

@@ -10,7 +10,7 @@ canal.conf:
   mode: tcp # kafka rocketMQ
   canalServerHost: 127.0.0.1:11111
 #  zookeeperHosts: slave1:2181
-#  mqServers: 127.0.0.1:9092 #or rocketmq
+#  mqServers: 127.0.0.1:9092 #or rocketmq nameservers
 #  flatMessage: true
   batchSize: 500
   syncBatchSize: 1000
@@ -18,6 +18,11 @@ canal.conf:
   timeout:
   accessKey:
   secretKey:
+# enableMessageTrace:
+# accessChannel:
+# customizedTraceTopic:
+# namespace:
+
 #  srcDataSources:
 #    defaultDS:
 #      url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true

client-adapter/launcher/src/main/resources/logback.xml

@@ -13,12 +13,12 @@
             <DefaultValue>adapter</DefaultValue>
         </discriminator>
 		<sift>
-			<appender name="FILE-${adapter}" class="ch.qos.logback.core.rolling.RollingFileAppender">
+			<appender name="FILE-adapter" class="ch.qos.logback.core.rolling.RollingFileAppender">
-				<File>../logs/${adapter}/${adapter}.log</File>
+				<File>../logs/adapter/adapter.log</File>
 				<rollingPolicy
 						class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
 					<!-- rollover daily -->
-					<fileNamePattern>../logs/${adapter}/%d{yyyy-MM-dd}/${adapter}-%d{yyyy-MM-dd}-%i.log.gz</fileNamePattern>
+					<fileNamePattern>../logs/adapter/%d{yyyy-MM-dd}/adapter-%d{yyyy-MM-dd}-%i.log.gz</fileNamePattern>
 					<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
 						<!-- or whenever the file size reaches 100MB -->
 						<maxFileSize>512MB</maxFileSize>
@@ -36,14 +36,22 @@
 
 	<logger name="com.alibaba.otter.canal.client.adapter.hbase" additivity="false">
 		<level value="DEBUG" />
+		<appender-ref ref="STDOUT"/>
 		<appender-ref ref="CANAL-ROOT" />
 	</logger>
 	<logger name="com.alibaba.otter.canal.client.adapter.es" additivity="false">
 		<level value="DEBUG" />
+		<appender-ref ref="STDOUT"/>
 		<appender-ref ref="CANAL-ROOT" />
 	</logger>
 	<logger name="com.alibaba.otter.canal.client.adapter.rdb" additivity="false">
 		<level value="DEBUG" />
+		<appender-ref ref="STDOUT"/>
+		<appender-ref ref="CANAL-ROOT" />
+	</logger>
+	<logger name="com.alibaba.otter.canal.client.adapter.logger" additivity="false">
+		<level value="DEBUG" />
+		<appender-ref ref="STDOUT"/>
 		<appender-ref ref="CANAL-ROOT" />
 	</logger>
 

client-adapter/logger/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.3-SNAPSHOT</version>
+        <version>1.1.4-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.alibaba.otter</groupId>

client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java

@@ -5,7 +5,6 @@ import java.util.Properties;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.slf4j.MDC;
 
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.serializer.SerializerFeature;
@@ -28,7 +27,6 @@ public class LoggerAdapterExample implements OuterAdapter {
 
     @Override
     public void init(OuterAdapterConfig configuration, Properties envProperties) {
-        MDC.put("adapter", "logger");
     }
 
     public void sync(List<Dml> dmls) {
@@ -43,6 +41,5 @@ public class LoggerAdapterExample implements OuterAdapter {
 
     @Override
     public void destroy() {
-        MDC.remove("adapter");
     }
 }

client-adapter/pom.xml

@@ -3,7 +3,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.alibaba.otter</groupId>
     <artifactId>canal.client-adapter</artifactId>
-    <version>1.1.3-SNAPSHOT</version>
+    <version>1.1.4-SNAPSHOT</version>
     <packaging>pom</packaging>
     <name>canal client adapter module for otter ${project.version}</name>
 

client-adapter/rdb/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.3-SNAPSHOT</version>
+        <version>1.1.4-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.alibaba.otter</groupId>

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

@@ -8,13 +8,10 @@ import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.ConcurrentHashMap;
 
-import javax.sql.DataSource;
-
 import org.apache.commons.lang.BooleanUtils;
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.slf4j.MDC;
 
 import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.otter.canal.client.adapter.OuterAdapter;
@@ -26,12 +23,7 @@ import com.alibaba.otter.canal.client.adapter.rdb.service.RdbEtlService;
 import com.alibaba.otter.canal.client.adapter.rdb.service.RdbMirrorDbSyncService;
 import com.alibaba.otter.canal.client.adapter.rdb.service.RdbSyncService;
 import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil;
-import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
+import com.alibaba.otter.canal.client.adapter.support.*;
-import com.alibaba.otter.canal.client.adapter.support.Dml;
-import com.alibaba.otter.canal.client.adapter.support.EtlResult;
-import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
-import com.alibaba.otter.canal.client.adapter.support.SPI;
-import com.alibaba.otter.canal.client.adapter.support.Util;
 
 /**
  * RDB适配器实现类
@@ -76,14 +68,13 @@ public class RdbAdapter implements OuterAdapter {
      */
     @Override
     public void init(OuterAdapterConfig configuration, Properties envProperties) {
-        MDC.put("adapter", "rdb");
         this.envProperties = envProperties;
         Map<String, MappingConfig> rdbMappingTmp = ConfigLoader.load(envProperties);
         // 过滤不匹配的key的配置
         rdbMappingTmp.forEach((key, mappingConfig) -> {
             if ((mappingConfig.getOuterAdapterKey() == null && configuration.getKey() == null)
-                || (mappingConfig.getOuterAdapterKey() != null && mappingConfig.getOuterAdapterKey()
+                || (mappingConfig.getOuterAdapterKey() != null
-                    .equalsIgnoreCase(configuration.getKey()))) {
+                    && mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) {
                 rdbMapping.put(key, mappingConfig);
             }
         });
@@ -143,8 +134,8 @@ public class RdbAdapter implements OuterAdapter {
         String threads = properties.get("threads");
         // String commitSize = properties.get("commitSize");
 
-        boolean skipDupException = BooleanUtils.toBoolean(configuration.getProperties()
+        boolean skipDupException = BooleanUtils
-            .getOrDefault("skipDupException", "true"));
+            .toBoolean(configuration.getProperties().getOrDefault("skipDupException", "true"));
         rdbSyncService = new RdbSyncService(dataSource,
             threads != null ? Integer.valueOf(threads) : null,
             skipDupException);
@@ -188,27 +179,16 @@ public class RdbAdapter implements OuterAdapter {
     public EtlResult etl(String task, List<String> params) {
         EtlResult etlResult = new EtlResult();
         MappingConfig config = rdbMapping.get(task);
+        RdbEtlService rdbEtlService = new RdbEtlService(dataSource, config);
         if (config != null) {
-            DataSource srcDataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
+                return rdbEtlService.importData(params);
-            if (srcDataSource != null) {
-                return RdbEtlService.importData(srcDataSource, dataSource, config, params);
-            } else {
-                etlResult.setSucceeded(false);
-                etlResult.setErrorMessage("DataSource not found");
-                return etlResult;
-            }
         } else {
             StringBuilder resultMsg = new StringBuilder();
             boolean resSucc = true;
-            // ds不为空说明传入的是destination
             for (MappingConfig configTmp : rdbMapping.values()) {
                 // 取所有的destination为task的配置
                 if (configTmp.getDestination().equals(task)) {
-                    DataSource srcDataSource = DatasourceConfig.DATA_SOURCES.get(configTmp.getDataSourceKey());
+                    EtlResult etlRes = rdbEtlService.importData(params);
-                    if (srcDataSource == null) {
-                        continue;
-                    }
-                    EtlResult etlRes = RdbEtlService.importData(srcDataSource, dataSource, configTmp, params);
                     if (!etlRes.getSucceeded()) {
                         resSucc = false;
                         resultMsg.append(etlRes.getErrorMessage()).append("\n");
@@ -304,6 +284,5 @@ public class RdbAdapter implements OuterAdapter {
         if (dataSource != null) {
             dataSource.close();
         }
-        MDC.remove("adapter");
     }
 }

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

@@ -3,6 +3,7 @@ package com.alibaba.otter.canal.client.adapter.rdb.config;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
+import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
 import org.apache.commons.lang.StringUtils;
 
 /**
@@ -11,7 +12,7 @@ import org.apache.commons.lang.StringUtils;
  * @author rewerma 2018-11-07 下午02:41:34
  * @version 1.0.0
  */
-public class MappingConfig {
+public class MappingConfig implements AdapterConfig {
 
     private String    dataSourceKey;      // 数据源key
 
@@ -73,6 +74,10 @@ public class MappingConfig {
         this.destination = destination;
     }
 
+    public AdapterMapping getMapping() {
+        return dbMapping;
+    }
+
     public void validate() {
         if (dbMapping.database == null || dbMapping.database.isEmpty()) {
             throw new NullPointerException("dbMapping.database");
@@ -85,7 +90,7 @@ public class MappingConfig {
         }
     }
 
-    public static class DbMapping {
+    public static class DbMapping implements AdapterMapping {
 
         private boolean             mirrorDb    = false;                 // 是否镜像库
         private String              database;                            // 数据库名或schema名

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

@@ -5,22 +5,17 @@ import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
 import javax.sql.DataSource;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
 import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig.DbMapping;
 import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil;
+import com.alibaba.otter.canal.client.adapter.support.AbstractEtlService;
+import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
 import com.alibaba.otter.canal.client.adapter.support.EtlResult;
 import com.alibaba.otter.canal.client.adapter.support.Util;
-import com.google.common.base.Joiner;
 
 /**
  * RDB ETL 操作业务类
@@ -28,139 +23,33 @@ import com.google.common.base.Joiner;
  * @author rewerma @ 2018-11-7
  * @version 1.0.0
  */
-public class RdbEtlService {
+public class RdbEtlService extends AbstractEtlService {
 
-    private static final Logger logger = LoggerFactory.getLogger(RdbEtlService.class);
+    private DataSource    targetDS;
+    private MappingConfig config;
+
+    public RdbEtlService(DataSource targetDS, MappingConfig config){
+        super("RDB", config);
+        this.targetDS = targetDS;
+        this.config = config;
+    }
 
     /**
      * 导入数据
      */
-    public static EtlResult importData(DataSource srcDS, DataSource targetDS, MappingConfig config,
+    public EtlResult importData(List<String> params) {
-                                       List<String> params) {
-        EtlResult etlResult = new EtlResult();
-        AtomicLong successCount = new AtomicLong();
-        List<String> errMsg = new ArrayList<>();
-        String hbaseTable = "";
-        try {
-            if (config == null) {
-                logger.error("Config is null!");
-                etlResult.setSucceeded(false);
-                etlResult.setErrorMessage("Config is null!");
-                return etlResult;
-            }
         DbMapping dbMapping = config.getDbMapping();
-
+        String sql = "SELECT * FROM " + dbMapping.getDatabase() + "." + dbMapping.getTable();
-            long start = System.currentTimeMillis();
+        return importData(sql, params);
-
-            // 拼接sql
-            StringBuilder sql = new StringBuilder(
-                "SELECT * FROM " + dbMapping.getDatabase() + "." + dbMapping.getTable());
-
-            // 拼接条件
-            appendCondition(params, dbMapping, srcDS, sql);
-
-            // 获取总数
-            String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT ";
-            long cnt = (Long) Util.sqlRS(srcDS, countSql, rs -> {
-                Long count = null;
-                try {
-                    if (rs.next()) {
-                        count = ((Number) rs.getObject(1)).longValue();
-                    }
-                } catch (Exception e) {
-                    logger.error(e.getMessage(), e);
-                }
-                return count == null ? 0 : count;
-            });
-
-            // 当大于1万条记录时开启多线程
-            if (cnt >= 10000) {
-                int threadCount = 3;
-                long perThreadCnt = cnt / threadCount;
-                ExecutorService executor = Util.newFixedThreadPool(threadCount, 5000L);
-                for (int i = 0; i < threadCount; i++) {
-                    long offset = i * perThreadCnt;
-                    Long size = null;
-                    if (i != threadCount - 1) {
-                        size = perThreadCnt;
-                    }
-                    String sqlFinal;
-                    if (size != null) {
-                        sqlFinal = sql + " LIMIT " + offset + "," + size;
-                    } else {
-                        sqlFinal = sql + " LIMIT " + offset + "," + cnt;
-                    }
-                    executor
-                        .execute(() -> executeSqlImport(srcDS, targetDS, sqlFinal, dbMapping, successCount, errMsg));
-                }
-
-                executor.shutdown();
-                while (!executor.awaitTermination(3, TimeUnit.SECONDS)) {
-                    // ignore
-                }
-            } else {
-                executeSqlImport(srcDS, targetDS, sql.toString(), dbMapping, successCount, errMsg);
-            }
-
-            logger.info(
-                dbMapping.getTable() + " etl completed in: " + (System.currentTimeMillis() - start) / 1000 + "s!");
-
-            etlResult
-                .setResultMessage("导入目标表 " + SyncUtil.getDbTableName(dbMapping) + " 数据：" + successCount.get() + " 条");
-        } catch (Exception e) {
-            logger.error(e.getMessage(), e);
-            errMsg.add(hbaseTable + " etl failed! ==>" + e.getMessage());
-        }
-
-        if (errMsg.isEmpty()) {
-            etlResult.setSucceeded(true);
-        } else {
-            etlResult.setErrorMessage(Joiner.on("\n").join(errMsg));
-        }
-        return etlResult;
-    }
-
-    private static void appendCondition(List<String> params, DbMapping dbMapping, DataSource ds,
-                                        StringBuilder sql) throws SQLException {
-        if (params != null && params.size() == 1 && dbMapping.getEtlCondition() == null) {
-            AtomicBoolean stExists = new AtomicBoolean(false);
-            // 验证是否有SYS_TIME字段
-            Util.sqlRS(ds, sql.toString(), rs -> {
-                try {
-                    ResultSetMetaData rsmd = rs.getMetaData();
-                    int cnt = rsmd.getColumnCount();
-                    for (int i = 1; i <= cnt; i++) {
-                        String columnName = rsmd.getColumnName(i);
-                        if ("SYS_TIME".equalsIgnoreCase(columnName)) {
-                            stExists.set(true);
-                            break;
-                        }
-                    }
-                } catch (Exception e) {
-                    // ignore
-                }
-                return null;
-            });
-            if (stExists.get()) {
-                sql.append(" WHERE SYS_TIME >= '").append(params.get(0)).append("' ");
-            }
-        } else if (dbMapping.getEtlCondition() != null && params != null) {
-            String etlCondition = dbMapping.getEtlCondition();
-            int size = params.size();
-            for (int i = 0; i < size; i++) {
-                etlCondition = etlCondition.replace("{" + i + "}", params.get(i));
-            }
-
-            sql.append(" ").append(etlCondition);
-        }
     }
 
     /**
      * 执行导入
      */
-    private static boolean executeSqlImport(DataSource srcDS, DataSource targetDS, String sql, DbMapping dbMapping,
+    protected boolean executeSqlImport(DataSource srcDS, String sql, List<Object> values,
-                                            AtomicLong successCount, List<String> errMsg) {
+                                       AdapterConfig.AdapterMapping mapping, AtomicLong impCount, List<String> errMsg) {
         try {
+            DbMapping dbMapping = (DbMapping) mapping;
             Map<String, String> columnsMap = new LinkedHashMap<>();
             Map<String, Integer> columnType = new LinkedHashMap<>();
 
@@ -183,18 +72,12 @@ public class RdbEtlService {
                 }
             });
 
-            Util.sqlRS(srcDS, sql, rs -> {
+            Util.sqlRS(srcDS, sql, values, rs -> {
                 int idx = 1;
 
                 try {
                     boolean completed = false;
 
-                    // if (dbMapping.isMapAll()) {
-                    // columnsMap = dbMapping.getAllColumns();
-                    // } else {
-                    // columnsMap = dbMapping.getTargetColumns();
-                    // }
-
                     StringBuilder insertSql = new StringBuilder();
                     insertSql.append("INSERT INTO ").append(SyncUtil.getDbTableName(dbMapping)).append(" (");
                     columnsMap
@@ -218,13 +101,13 @@ public class RdbEtlService {
                             pstmt.clearParameters();
 
                             // 删除数据
-                            Map<String, Object> values = new LinkedHashMap<>();
+                            Map<String, Object> pkVal = new LinkedHashMap<>();
                             StringBuilder deleteSql = new StringBuilder(
                                 "DELETE FROM " + SyncUtil.getDbTableName(dbMapping) + " WHERE ");
-                            appendCondition(dbMapping, deleteSql, values, rs);
+                            appendCondition(dbMapping, deleteSql, pkVal, rs);
                             try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) {
                                 int k = 1;
-                                for (Object val : values.values()) {
+                                for (Object val : pkVal.values()) {
                                     pstmt2.setObject(k++, val);
                                 }
                                 pstmt2.execute();
@@ -260,9 +143,9 @@ public class RdbEtlService {
                                 completed = true;
                             }
                             idx++;
-                            successCount.incrementAndGet();
+                            impCount.incrementAndGet();
                             if (logger.isDebugEnabled()) {
-                                logger.debug("successful import count:" + successCount.get());
+                                logger.debug("successful import count:" + impCount.get());
                             }
                         }
                         if (!completed) {

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

@@ -272,7 +272,7 @@ public class RdbSyncService {
             batchExecutor.execute(insertSql.toString(), values);
         } catch (SQLException e) {
             if (skipDupException
-                && (e.getMessage().contains("Duplicate entry") || e.getMessage().startsWith("ORA-00001: 违反唯一约束条件"))) {
+                && (e.getMessage().contains("Duplicate entry") || e.getMessage().startsWith("ORA-00001:"))) {
                 // ignore
                 // TODO 增加更多关系数据库的主键冲突的错误码
             } else {

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

@@ -4,7 +4,14 @@ import java.io.Reader;
 import java.io.StringReader;
 import java.math.BigDecimal;
 import java.nio.charset.StandardCharsets;
-import java.sql.*;
+import java.sql.Blob;
+import java.sql.Clob;
+import java.sql.Date;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.sql.Types;
 import java.util.Collection;
 import java.util.LinkedHashMap;
 import java.util.Map;

client-adapter/rdb/src/main/resources/rdb/mytest_user.yml

@@ -16,6 +16,8 @@ dbMapping:
     role_id:
     c_time:
     test1:
+  etlCondition: "where c_time>={}"
+  commitBatch: 3000 # 批量提交的大小
 
 
 ## Mirror schema synchronize config

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

@@ -7,10 +7,11 @@ import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
 
+import org.junit.Ignore;
 import org.junit.Test;
 
 import com.alibaba.druid.pool.DruidDataSource;
-
+@Ignore
 public class DBTest {
 
     @Test

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

@@ -1,13 +1,19 @@
 package com.alibaba.otter.canal.client.adapter.rdb.test.sync;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Date;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
 
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter;
 import com.alibaba.otter.canal.client.adapter.support.Dml;
-
+@Ignore
 public class OracleSyncTest {
 
     private RdbAdapter rdbAdapter;

client/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.1.3-SNAPSHOT</version>
+		<version>1.1.4-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>
@@ -106,8 +106,8 @@
 			<artifactId>rocketmq-client</artifactId>
 		</dependency>
 		<dependency>
-			<groupId>com.aliyun.openservices</groupId>
+			<groupId>org.apache.rocketmq</groupId>
-			<artifactId>aliware-apache-rocketmq-cloud</artifactId>
+			<artifactId>rocketmq-acl</artifactId>
 		</dependency>
 		<!-- 客户端要使用请单独引入kafka-clients依赖 -->
 		<dependency>

client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java

@@ -56,6 +56,7 @@ public class KafkaCanalConnector implements CanalMQConnector {
         properties.put("auto.offset.reset", "latest"); // 如果没有offset则从最后的offset开始读
         properties.put("request.timeout.ms", "40000"); // 必须大于session.timeout.ms的设置
         properties.put("session.timeout.ms", "30000"); // 默认为30秒
+        properties.put("isolation.level", "read_committed");
         if (batchSize == null) {
             batchSize = 100;
         }

client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java

@@ -6,6 +6,9 @@ import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.lang.StringUtils;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
+import org.apache.rocketmq.client.AccessChannel;
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
 import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
@@ -24,8 +27,6 @@ import com.alibaba.otter.canal.client.impl.SimpleCanalConnector;
 import com.alibaba.otter.canal.protocol.FlatMessage;
 import com.alibaba.otter.canal.protocol.Message;
 import com.alibaba.otter.canal.protocol.exception.CanalClientException;
-import com.aliyun.openservices.apache.api.impl.authority.SessionCredentials;
-import com.aliyun.openservices.apache.api.impl.rocketmq.ClientRPCHook;
 import com.google.common.collect.Lists;
 
 /**
@@ -41,6 +42,7 @@ import com.google.common.collect.Lists;
 public class RocketMQCanalConnector implements CanalMQConnector {
 
     private static final Logger                 logger               = LoggerFactory.getLogger(RocketMQCanalConnector.class);
+    private static final String                 CLOUD_ACCESS_CHANNEL = "cloud";
 
     private String                              nameServer;
     private String                              topic;
@@ -54,6 +56,26 @@ public class RocketMQCanalConnector implements CanalMQConnector {
     private volatile ConsumerBatchMessage       lastGetBatchMessage = null;
     private String                              accessKey;
     private String                              secretKey;
+    private String                              customizedTraceTopic;
+    private boolean                             enableMessageTrace = false;
+    private String                              accessChannel;
+    private String                              namespace;
+
+    public RocketMQCanalConnector(String nameServer, String topic, String groupName, String accessKey,
+        String secretKey, Integer batchSize, boolean flatMessage, boolean enableMessageTrace,
+        String customizedTraceTopic, String accessChannel, String namespace) {
+        this(nameServer, topic, groupName, accessKey, secretKey, batchSize, flatMessage, enableMessageTrace, customizedTraceTopic, accessChannel);
+        this.namespace = namespace;
+    }
+
+    public RocketMQCanalConnector(String nameServer, String topic, String groupName, String accessKey,
+        String secretKey, Integer batchSize, boolean flatMessage, boolean enableMessageTrace,
+        String customizedTraceTopic, String accessChannel) {
+        this(nameServer, topic, groupName, accessKey, secretKey, batchSize, flatMessage);
+        this.enableMessageTrace = enableMessageTrace;
+        this.customizedTraceTopic = customizedTraceTopic;
+        this.accessChannel = accessChannel;
+    }
 
     public RocketMQCanalConnector(String nameServer, String topic, String groupName, Integer batchSize,
                                   boolean flatMessage){
@@ -73,16 +95,24 @@ public class RocketMQCanalConnector implements CanalMQConnector {
     }
 
     public void connect() throws CanalClientException {
-
         RPCHook rpcHook = null;
         if (null != accessKey && accessKey.length() > 0 && null != secretKey && secretKey.length() > 0) {
             SessionCredentials sessionCredentials = new SessionCredentials();
             sessionCredentials.setAccessKey(accessKey);
             sessionCredentials.setSecretKey(secretKey);
-            rpcHook = new ClientRPCHook(sessionCredentials);
+            rpcHook = new AclClientRPCHook(sessionCredentials);
         }
-        rocketMQConsumer = new DefaultMQPushConsumer(groupName, rpcHook, new AllocateMessageQueueAveragely());
+
+        rocketMQConsumer = new DefaultMQPushConsumer(groupName, rpcHook, new AllocateMessageQueueAveragely(), enableMessageTrace, customizedTraceTopic);
         rocketMQConsumer.setVipChannelEnabled(false);
+        if (CLOUD_ACCESS_CHANNEL.equals(this.accessChannel)) {
+            rocketMQConsumer.setAccessChannel(AccessChannel.CLOUD);
+        }
+
+        if (!StringUtils.isEmpty(this.namespace)) {
+            rocketMQConsumer.setNamespace(this.namespace);
+        }
+
         if (!StringUtils.isBlank(nameServer)) {
             rocketMQConsumer.setNamesrvAddr(nameServer);
         }
@@ -131,7 +161,9 @@ public class RocketMQCanalConnector implements CanalMQConnector {
     }
 
     private boolean process(List<MessageExt> messageExts) {
-        logger.info("Get Message:{}", messageExts);
+        if (logger.isDebugEnabled()) {
+            logger.debug("Get Message: {}", messageExts);
+        }
         List messageList = Lists.newArrayList();
         for (MessageExt messageExt : messageExts) {
             byte[] data = messageExt.getBody();

client/src/test/java/com/alibaba/otter/canal/client/running/ClientRunningTest.java

@@ -9,6 +9,7 @@ import org.apache.commons.lang.math.RandomUtils;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import com.alibaba.otter.canal.client.impl.running.ClientRunningData;
@@ -17,7 +18,7 @@ import com.alibaba.otter.canal.client.impl.running.ClientRunningMonitor;
 import com.alibaba.otter.canal.common.utils.AddressUtils;
 import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
 import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils;
-
+@Ignore
 public class ClientRunningTest extends AbstractZkTest {
 
     private ZkClientx zkclientx = new ZkClientx(cluster1 + ";" + cluster2);

common/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.1.3-SNAPSHOT</version>
+		<version>1.1.4-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<artifactId>canal.common</artifactId>

common/src/test/java/com/alibaba/otter/canal/common/ServerRunningTest.java

@@ -8,6 +8,7 @@ import org.apache.commons.lang.math.RandomUtils;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
@@ -15,7 +16,7 @@ import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils;
 import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningData;
 import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningListener;
 import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitor;
-
+@Ignore
 public class ServerRunningTest extends AbstractZkTest {
 
     private ZkClientx zkclientx = new ZkClientx(cluster1 + ";" + cluster2);

dbsync/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.1.3-SNAPSHOT</version>
+		<version>1.1.4-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/QueryLogEvent.java

@@ -706,10 +706,8 @@ public class QueryLogEvent extends LogEvent {
                          * That's why you must write status vars in growing
                          * order of code
                          */
-                        if (logger.isDebugEnabled()) {
+                        logger.error("Query_log_event has unknown status vars (first has code: " + code
-                            logger.debug("Query_log_event has unknown status vars (first has code: " + code
                                      + "), skipping the rest of them");
-                        }
                         break; // Break loop
                 }
             }

dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogEvent.java

@@ -2,6 +2,7 @@ package com.taobao.tddl.dbsync.binlog.event;
 
 import java.util.BitSet;
 
+import com.taobao.tddl.dbsync.binlog.exception.TableIdNotFoundException;
 import com.taobao.tddl.dbsync.binlog.LogBuffer;
 import com.taobao.tddl.dbsync.binlog.LogContext;
 import com.taobao.tddl.dbsync.binlog.LogEvent;
@@ -179,6 +180,10 @@ public abstract class RowsLogEvent extends LogEvent {
     public final void fillTable(LogContext context) {
         table = context.getTable(tableId);
 
+        if (table == null) {
+            throw new TableIdNotFoundException("not found tableId:" + tableId);
+        }
+
         // end of statement check:
         if ((flags & RowsLogEvent.STMT_END_F) != 0) {
             // Now is safe to clear ignored map (clear_tables will also

dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/exception/TableIdNotFoundException.java

@@ -1,4 +1,4 @@
-package com.alibaba.otter.canal.parse.exception;
+package com.taobao.tddl.dbsync.binlog.exception;
 
 import com.alibaba.otter.canal.common.CanalException;
 

dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/DirectLogFetcherTest.java

@@ -6,6 +6,7 @@ import java.sql.DriverManager;
 import java.sql.Statement;
 
 import org.junit.Assert;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import com.taobao.tddl.dbsync.binlog.event.DeleteRowsLogEvent;
@@ -16,7 +17,7 @@ import com.taobao.tddl.dbsync.binlog.event.UpdateRowsLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.XidLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.mariadb.AnnotateRowsEvent;
-
+@Ignore
 public class DirectLogFetcherTest extends BaseLogFetcherTest {
 
     @Test

dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/FileLogFetcherTest.java

@@ -6,6 +6,7 @@ import java.net.URL;
 
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import com.taobao.tddl.dbsync.binlog.event.DeleteRowsLogEvent;
@@ -16,7 +17,7 @@ import com.taobao.tddl.dbsync.binlog.event.UpdateRowsLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.XidLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.mariadb.AnnotateRowsEvent;
-
+@Ignore
 public class FileLogFetcherTest extends BaseLogFetcherTest {
 
     private String directory;

deployer/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.1.3-SNAPSHOT</version>
+		<version>1.1.4-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java

@@ -48,11 +48,17 @@ public class CanalConstants {
     public static final String CANAL_MQ_FLATMESSAGE                 = ROOT + "." + "mq.flatMessage";
     public static final String CANAL_MQ_COMPRESSION_TYPE            = ROOT + "." + "mq.compressionType";
     public static final String CANAL_MQ_ACKS                        = ROOT + "." + "mq.acks";
-    public static final String CANAL_MQ_TRANSACTION              = ROOT + "." + "mq.transaction";
     public static final String CANAL_MQ_PRODUCERGROUP               = ROOT + "." + "mq.producerGroup";
     public static final String CANAL_ALIYUN_ACCESSKEY               = ROOT + "." + "aliyun.accessKey";
     public static final String CANAL_ALIYUN_SECRETKEY               = ROOT + "." + "aliyun.secretKey";
     public static final String CANAL_MQ_PROPERTIES                  = ROOT + "." + "mq.properties";
+    public static final String CANAL_MQ_ENABLE_MESSAGE_TRACE        = ROOT + "." + "mq.enableMessageTrace";
+    public static final String CANAL_MQ_ACCESS_CHANNEL              = ROOT + "." + "mq.accessChannel";
+    public static final String CANAL_MQ_CUSTOMIZED_TRACE_TOPIC      = ROOT + "." + "mq.customizedTraceTopic";
+    public static final String CANAL_MQ_NAMESPACE                   = ROOT + "." + "mq.namespace";
+    public static final String CANAL_MQ_KAFKA_KERBEROS_ENABLE       = ROOT + "." + "mq.kafka.kerberos.enable";
+    public static final String CANAL_MQ_KAFKA_KERBEROS_KRB5FILEPATH = ROOT + "." + "mq.kafka.kerberos.krb5FilePath";
+    public static final String CANAL_MQ_KAFKA_KERBEROS_JAASFILEPATH = ROOT + "." + "mq.kafka.kerberos.jaasFilePath";
 
     public static String getInstanceModeKey(String destination) {
         return MessageFormat.format(INSTANCE_MODE_TEMPLATE, destination);

deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java

@@ -152,6 +152,9 @@ public class CanalController {
                             try {
                                 MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
                                 embededCanalServer.start(destination);
+                                if (canalMQStarter != null) {
+                                    canalMQStarter.startDestination(destination);
+                                }
                             } finally {
                                 MDC.remove(CanalConstants.MDC_DESTINATION);
                             }
@@ -160,6 +163,9 @@ public class CanalController {
                         public void processActiveExit() {
                             try {
                                 MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
+                                if (canalMQStarter != null) {
+                                    canalMQStarter.stopDestination(destination);
+                                }
                                 embededCanalServer.stop(destination);
                             } finally {
                                 MDC.remove(CanalConstants.MDC_DESTINATION);
@@ -234,9 +240,6 @@ public class CanalController {
                         ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
                         if (!config.getLazy() && !runningMonitor.isStart()) {
                             runningMonitor.start();
-                            if (canalMQStarter != null) {
-                                canalMQStarter.startDestination(destination);
-                            }
                         }
                     }
                 }
@@ -245,9 +248,6 @@ public class CanalController {
                     // 此处的stop，代表强制退出，非HA机制，所以需要退出HA的monitor和配置信息
                     InstanceConfig config = instanceConfigs.remove(destination);
                     if (config != null) {
-                        if (canalMQStarter != null) {
-                            canalMQStarter.stopDestination(destination);
-                        }
                         embededCanalServer.stop(destination);
                         ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
                         if (runningMonitor.isStart()) {

deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java

@@ -52,6 +52,8 @@ public class CanalStater {
             // disable netty
             System.setProperty(CanalConstants.CANAL_WITHOUT_NETTY, "true");
             String autoScan = CanalController.getProperty(properties, CanalConstants.CANAL_AUTO_SCAN);
+            // 设置为raw避免ByteString->Entry的二次解析
+            System.setProperty("canal.instance.memory.rawEntry", "false");
             if ("true".equals(autoScan)) {
                 String rootDir = CanalController.getProperty(properties, CanalConstants.CANAL_CONF_DIR);
                 if (StringUtils.isEmpty(rootDir)) {
@@ -63,7 +65,8 @@ public class CanalStater {
 
                         public boolean accept(File pathname) {
                             String filename = pathname.getName();
-                            return pathname.isDirectory() && !"spring".equalsIgnoreCase(filename);
+                            return pathname.isDirectory() && !"spring".equalsIgnoreCase(filename)
+                                   && !"metrics".equalsIgnoreCase(filename);
                         }
                     });
                     if (instanceDirs != null && instanceDirs.length > 0) {
@@ -194,16 +197,52 @@ public class CanalStater {
         if (!StringUtils.isEmpty(aliyunSecretKey)) {
             mqProperties.setAliyunSecretKey(aliyunSecretKey);
         }
-        String transaction = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_TRANSACTION);
-        if (!StringUtils.isEmpty(transaction)) {
-            mqProperties.setTransaction(Boolean.valueOf(transaction));
-        }
 
         String producerGroup = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_PRODUCERGROUP);
         if (!StringUtils.isEmpty(producerGroup)) {
             mqProperties.setProducerGroup(producerGroup);
         }
 
+        String enableMessageTrace = CanalController.getProperty(properties,
+            CanalConstants.CANAL_MQ_ENABLE_MESSAGE_TRACE);
+        if (!StringUtils.isEmpty(enableMessageTrace)) {
+            mqProperties.setEnableMessageTrace(Boolean.valueOf(enableMessageTrace));
+        }
+
+        String accessChannel = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_ACCESS_CHANNEL);
+        if (!StringUtils.isEmpty(accessChannel)) {
+            mqProperties.setAccessChannel(accessChannel);
+        }
+
+        String customizedTraceTopic = CanalController.getProperty(properties,
+            CanalConstants.CANAL_MQ_CUSTOMIZED_TRACE_TOPIC);
+        if (!StringUtils.isEmpty(customizedTraceTopic)) {
+            mqProperties.setCustomizedTraceTopic(customizedTraceTopic);
+        }
+
+        String namespace = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_NAMESPACE);
+        if (!StringUtils.isEmpty(namespace)) {
+            mqProperties.setNamespace(namespace);
+        }
+
+        String kafkaKerberosEnable = CanalController.getProperty(properties,
+            CanalConstants.CANAL_MQ_KAFKA_KERBEROS_ENABLE);
+        if (!StringUtils.isEmpty(kafkaKerberosEnable)) {
+            mqProperties.setKerberosEnable(Boolean.valueOf(kafkaKerberosEnable));
+        }
+
+        String kafkaKerberosKrb5Filepath = CanalController.getProperty(properties,
+            CanalConstants.CANAL_MQ_KAFKA_KERBEROS_KRB5FILEPATH);
+        if (!StringUtils.isEmpty(kafkaKerberosKrb5Filepath)) {
+            mqProperties.setKerberosKrb5FilePath(kafkaKerberosKrb5Filepath);
+        }
+
+        String kafkaKerberosJaasFilepath = CanalController.getProperty(properties,
+            CanalConstants.CANAL_MQ_KAFKA_KERBEROS_JAASFILEPATH);
+        if (!StringUtils.isEmpty(kafkaKerberosJaasFilepath)) {
+            mqProperties.setKerberosJaasFilePath(kafkaKerberosJaasFilepath);
+        }
+
         for (Object key : properties.keySet()) {
             key = StringUtils.trim(key.toString());
             if (((String) key).startsWith(CanalConstants.CANAL_MQ_PROPERTIES)) {

deployer/src/main/resources/canal.properties

@@ -108,13 +108,23 @@ canal.mq.servers = 127.0.0.1:6667
 canal.mq.retries = 0
 canal.mq.batchSize = 16384
 canal.mq.maxRequestSize = 1048576
-canal.mq.lingerMs = 1
+canal.mq.lingerMs = 100
 canal.mq.bufferMemory = 33554432
 canal.mq.canalBatchSize = 50
 canal.mq.canalGetTimeout = 100
 canal.mq.flatMessage = true
 canal.mq.compressionType = none
 canal.mq.acks = all
-# use transaction for kafka flatMessage batch produce
-canal.mq.transaction = false
 #canal.mq.properties. =
+canal.mq.producerGroup = test
+# Set this value to "cloud", if you want open message trace feature in aliyun.
+canal.mq.accessChannel = local
+# aliyun mq namespace
+#canal.mq.namespace =
+
+##################################################
+#########     Kafka Kerberos Info    #############
+##################################################
+canal.mq.kafka.kerberos.enable = false
+canal.mq.kafka.kerberos.krb5FilePath = "../conf/kerberos/krb5.conf"
+canal.mq.kafka.kerberos.jaasFilePath = "../conf/kerberos/jaas.conf"

deployer/src/main/resources/example/instance.properties

@@ -33,7 +33,6 @@ canal.instance.tsdb.enable=true
 canal.instance.dbUsername=canal
 canal.instance.dbPassword=canal
 canal.instance.connectionCharset = UTF-8
-canal.instance.defaultDatabaseName =test
 # enable druid Decrypt database password
 canal.instance.enableDruid=false
 #canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==
@@ -46,9 +45,9 @@ canal.instance.filter.black.regex=
 # mq config
 canal.mq.topic=example
 # dynamic topic route by schema or table regex
-#canal.mq.dynamicTopic=.*,mytest,mytest\\..*,mytest2.user
+#canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
 canal.mq.partition=0
 # hash partition config
-#canal.mq.partitionsNum=4
+#canal.mq.partitionsNum=3
 #canal.mq.partitionHash=test.table:id^name,.*\\..*
 #################################################

deployer/src/main/resources/spring/default-instance.xml

@@ -136,7 +136,7 @@
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 		<property name="standbyInfo">
@@ -146,7 +146,7 @@
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 

deployer/src/main/resources/spring/file-instance.xml

@@ -121,7 +121,7 @@
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 		<property name="standbyInfo">
@@ -131,7 +131,7 @@
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 

deployer/src/main/resources/spring/group-instance.xml

@@ -118,7 +118,7 @@
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 		<property name="standbyInfo">
@@ -128,7 +128,7 @@
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 
@@ -220,7 +220,7 @@
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 		<property name="standbyInfo">
@@ -230,7 +230,7 @@
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 

deployer/src/main/resources/spring/memory-instance.xml

@@ -109,7 +109,7 @@
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 		<property name="standbyInfo">
@@ -119,7 +119,7 @@
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 

deployer/src/main/resources/spring/tsdb/h2-tsdb.xml

@@ -43,6 +43,7 @@
         <property name="testOnBorrow" value="false" />
         <property name="testOnReturn" value="false" />
         <property name="useUnfairLock" value="true" />
+        <property name="validationQuery" value="SELECT 1" />
 	</bean>
 
     <bean id="sqlMapClient" class="org.springframework.orm.ibatis.SqlMapClientFactoryBean">

driver/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.1.3-SNAPSHOT</version>
+		<version>1.1.4-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java

@@ -2,19 +2,25 @@ package com.alibaba.otter.canal.parse.driver.mysql;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.security.DigestException;
+import java.security.NoSuchAlgorithmException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket;
+import com.alibaba.otter.canal.parse.driver.mysql.packets.client.AuthSwitchResponsePacket;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.client.ClientAuthenticationPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.client.QuitCommandPacket;
+import com.alibaba.otter.canal.parse.driver.mysql.packets.server.AuthSwitchRequestMoreData;
+import com.alibaba.otter.canal.parse.driver.mysql.packets.server.AuthSwitchRequestPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ErrorPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.server.HandshakeInitializationPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.server.Reply323Packet;
 import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel;
 import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannelPool;
+import com.alibaba.otter.canal.parse.driver.mysql.utils.MSC;
 import com.alibaba.otter.canal.parse.driver.mysql.utils.MySQLPasswordEncrypter;
 import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
 
@@ -32,7 +38,7 @@ public class MysqlConnector {
     private String              password;
 
     private byte                charsetNumber     = 33;
-    private String              defaultSchema     = "test";
+    private String              defaultSchema;
     private int                 soTimeout         = 30 * 1000;
     private int                 connTimeout       = 5 * 1000;
     private int                 receiveBufferSize = 16 * 1024;
@@ -150,6 +156,7 @@ public class MysqlConnector {
     }
 
     private void negotiate(SocketChannel channel) throws IOException {
+        // https://dev.mysql.com/doc/internals/en/connection-phase-packets.html#packet-Protocol
         HeaderPacket header = PacketManager.readHeader(channel, 4, timeout);
         byte[] body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
         if (body[0] < 0) {// check field_count
@@ -165,10 +172,14 @@ public class MysqlConnector {
         }
         HandshakeInitializationPacket handshakePacket = new HandshakeInitializationPacket();
         handshakePacket.fromBytes(body);
+        if (handshakePacket.protocolVersion != MSC.DEFAULT_PROTOCOL_VERSION) {
+            // HandshakeV9
+            auth323(channel, (byte) (header.getPacketSequenceNumber() + 1), handshakePacket.seed);
+            return;
+        }
+
         connectionId = handshakePacket.threadId; // 记录一下connection
-
         logger.info("handshake initialization packet received, prepare the client authentication packet to send");
-
         ClientAuthenticationPacket clientAuth = new ClientAuthenticationPacket();
         clientAuth.setCharsetNumber(charsetNumber);
 
@@ -177,6 +188,7 @@ public class MysqlConnector {
         clientAuth.setServerCapabilities(handshakePacket.serverCapabilities);
         clientAuth.setDatabaseName(defaultSchema);
         clientAuth.setScrumbleBuff(joinAndCreateScrumbleBuff(handshakePacket));
+        clientAuth.setAuthPluginName("mysql_native_password".getBytes());
 
         byte[] clientAuthPkgBody = clientAuth.toBytes();
         HeaderPacket h = new HeaderPacket();
@@ -192,15 +204,71 @@ public class MysqlConnector {
         body = null;
         body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
         assert body != null;
+        byte marker = body[0];
+        if (marker == -2 || marker == 1) {
+            byte[] authData = null;
+            String pluginName = null;
+            if (marker == 1) {
+                AuthSwitchRequestMoreData packet = new AuthSwitchRequestMoreData();
+                packet.fromBytes(body);
+                authData = packet.authData;
+            } else {
+                AuthSwitchRequestPacket packet = new AuthSwitchRequestPacket();
+                packet.fromBytes(body);
+                authData = packet.authData;
+                pluginName = packet.authName;
+            }
+
+            boolean isSha2Password = false;
+            byte[] encryptedPassword = null;
+            if (pluginName != null && "mysql_native_password".equals(pluginName)) {
+                try {
+                    encryptedPassword = MySQLPasswordEncrypter.scramble411(getPassword().getBytes(), authData);
+                } catch (NoSuchAlgorithmException e) {
+                    throw new RuntimeException("can't encrypt password that will be sent to MySQL server.", e);
+                }
+            } else if (pluginName != null && "caching_sha2_password".equals(pluginName)) {
+                isSha2Password = true;
+                try {
+                    encryptedPassword = MySQLPasswordEncrypter.scrambleCachingSha2(getPassword().getBytes(), authData);
+                } catch (DigestException e) {
+                    throw new RuntimeException("can't encrypt password that will be sent to MySQL server.", e);
+                }
+            }
+            assert encryptedPassword != null;
+            AuthSwitchResponsePacket responsePacket = new AuthSwitchResponsePacket();
+            responsePacket.authData = encryptedPassword;
+            byte[] auth = responsePacket.toBytes();
+
+            h = new HeaderPacket();
+            h.setPacketBodyLength(auth.length);
+            h.setPacketSequenceNumber((byte) (header.getPacketSequenceNumber() + 1));
+            PacketManager.writePkg(channel, h.toBytes(), auth);
+            logger.info("auth switch response packet is sent out.");
+
+            header = null;
+            header = PacketManager.readHeader(channel, 4);
+            body = null;
+            body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
+            assert body != null;
+            if (isSha2Password) {
+                if (body[0] == 0x01 && body[1] == 0x04) {
+                    // password auth failed
+                    throw new IOException("caching_sha2_password Auth failed");
+                }
+
+                header = null;
+                header = PacketManager.readHeader(channel, 4);
+                body = null;
+                body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
+            }
+        }
+
         if (body[0] < 0) {
             if (body[0] == -1) {
                 ErrorPacket err = new ErrorPacket();
                 err.fromBytes(body);
                 throw new IOException("Error When doing Client Authentication:" + err.toString());
-            } else if (body[0] == -2) {
-                auth323(channel, header.getPacketSequenceNumber(), handshakePacket.seed);
-                // throw new
-                // IOException("Unexpected EOF packet at Client Authentication.");
             } else {
                 throw new IOException("unpexpected packet with field_count=" + body[0]);
             }

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/Capability.java

@@ -0,0 +1,170 @@
+package com.alibaba.otter.canal.parse.driver.mysql.packets;
+
+/**
+ * https://dev.mysql.com/doc/internals/en/capability-flags.html#packet-Protocol
+ * ::CapabilityFlags
+ */
+public interface Capability {
+
+    // Use the improved version of Old Password Authentication.
+    // Assumed to be set since 4.1.1.
+    int CLIENT_LONG_PASSWORD                  = 0x00000001;
+
+    // Send found rows instead of affected rows in EOF_Packet.
+    int CLIENT_FOUND_ROWS                     = 0x00000002;
+
+    // https://dev.mysql.com/doc/internals/en/com-query-response.html#packet-Protocol::ColumnDefinition320
+    // Longer flags in Protocol::ColumnDefinition320.
+    // Server:Supports longer flags.
+    // Client:Expects longer flags.
+    // 执行查询sql时，除了返回结果集，还返回元数据
+    int CLIENT_LONG_FLAG                      = 0x00000004;
+
+    // 可以在handshake时，指定一个数据库名
+    // Database (schema) name can be specified on connect in Handshake Response
+    // Packet.
+    // Server: Supports schema-name in Handshake Response Packet.
+    // Client: Handshake Response Packet contains a schema-name.
+    int CLIENT_CONNECT_WITH_DB                = 0x00000008;
+
+    // Server: Do not permit database.table.column.
+    int CLIENT_NO_SCHEMA                      = 0x00000010;
+
+    // Compression protocol supported.
+    // Server:Supports compression.
+    // Client:Switches to Compression compressed protocol after successful
+    // authentication.
+    int CLIENT_COMPRESS                       = 0x00000020;
+
+    // Special handling of ODBC behavior.
+    // No special behavior since 3.22.
+    int CLIENT_ODBC                           = 0x00000040;
+
+    // Can use LOAD DATA LOCAL.
+    // Server:Enables the LOCAL INFILE request of LOAD DATA|XML.
+    // Client:Will handle LOCAL INFILE request.
+    int CLIENT_LOCAL_FILES                    = 0x00000080;
+
+    // Server: Parser can ignore spaces before '('.
+    // Client: Let the parser ignore spaces before '('.
+    int CLIENT_IGNORE_SPACE                   = 0x00000100;
+
+    // Server:Supports the 4.1 protocol,
+    // 4.1协议中，
+    // OKPacket将会包含warning count
+    // ERR_Packet包含SQL state
+    // EOF_Packet包含warning count和status flags
+    // Client:Uses the 4.1 protocol.
+    // Note: this value was CLIENT_CHANGE_USER in 3.22, unused in 4.0
+    // If CLIENT_PROTOCOL_41 is set：
+    // 1、the ok packet contains a warning count.
+    // https://dev.mysql.com/doc/internals/en/packet-OK_Packet.html
+    // 2、ERR_Packet It contains a SQL state value if CLIENT_PROTOCOL_41 is
+    // enabled. //https://dev.mysql.com/doc/internals/en/packet-ERR_Packet.html
+    // 3、EOF_Packet If CLIENT_PROTOCOL_41 is enabled, the EOF packet contains a
+    // warning count and status flags.
+    // https://dev.mysql.com/doc/internals/en/packet-EOF_Packet.html
+    int CLIENT_PROTOCOL_41                    = 0x00000200;
+
+    // wait_timeout versus wait_interactive_timeout.
+    // Server:Supports interactive and noninteractive clients.
+    // Client:Client is interactive.
+    int CLIENT_INTERACTIVE                    = 0x00000400;
+
+    // Server: Supports SSL.
+    // Client: Switch to SSL after sending the capability-flags.
+    int CLIENT_SSL                            = 0x00000800;
+
+    // Client: Do not issue SIGPIPE if network failures occur (libmysqlclient
+    // only).
+    int CLIENT_IGNORE_SIGPIPE                 = 0x00001000;
+
+    // Server: Can send status flags in EOF_Packet.
+    // Client:Expects status flags in EOF_Packet.
+    // Note:This flag is optional in 3.23, but always set by the server since
+    // 4.0.
+    int CLIENT_TRANSACTIONS                   = 0x00002000;
+
+    // Unused
+    // Note: Was named CLIENT_PROTOCOL_41 in 4.1.0.
+    int CLIENT_RESERVED                       = 0x00004000;
+
+    /**
+     * <pre>
+     *      服务端返回20 byte随机字节，客户端利用其对密码进行加密，加密算法如下：
+     *      https://dev.mysql.com/doc/internals/en/secure-password-authentication.html#packet-Authentication::Native41
+     *      Authentication::Native41:
+     *      client-side expects a 20-byte random challenge
+     *      client-side returns a 20-byte response based on the algorithm described later
+     *      Name
+     *      mysql_native_password
+     *      Requires
+     *      CLIENT_SECURE_CONNECTION
+     *      Image description follows.
+     *      Image description
+     *      This method fixes a 2 short-comings of the Old Password Authentication:
+     *      (https://dev.mysql.com/doc/internals/en/old-password-authentication.html#packet-Authentication::Old)
+     *      using a tested, crypto-graphic hashing function which isn't broken
+     *      knowning the content of the hash in the mysql.user table isn't enough to authenticate against the MySQL Server.
+     *      The password is calculated by:
+     *      SHA1( password ) XOR SHA1( "20-bytes random data from server" <concat> SHA1( SHA1( password ) ) )
+     * </pre>
+     */
+    int CLIENT_SECURE_CONNECTION              = 0x00008000;
+
+    // Server:Can handle multiple statements per COM_QUERY and COM_STMT_PREPARE.
+    // Client:May send multiple statements per COM_QUERY and COM_STMT_PREPARE.
+    // Note:Was named CLIENT_MULTI_QUERIES in 4.1.0, renamed later.
+    // Requires:CLIENT_PROTOCOL_41
+    int CLIENT_MULTI_STATEMENTS               = 0x00010000;
+
+    // Server: Can send multiple resultsets for COM_QUERY.
+    // Client: Can handle multiple resultsets for COM_QUERY.
+    // Requires:CLIENT_PROTOCOL_41
+    int CLIENT_MULTI_RESULTS                  = 0x00020000;
+
+    // Server: Can send multiple resultsets for ComStmtExecutePacket.
+    // Client: Can handle multiple resultsets for ComStmtExecutePacket.
+    // Requires:CLIENT_PROTOCOL_41
+    int CLIENT_PS_MULTI_RESULTS               = 0x00040000;
+
+    // Server:Sends extra data in Initial Handshake Packet and supports the
+    // pluggable authentication protocol.
+    // Client: Supports authentication plugins.
+    // Requires: CLIENT_PROTOCOL_41
+    int CLIENT_PLUGIN_AUTH                    = 0x00080000;
+
+    // Server: Permits connection attributes in Protocol::HandshakeResponse41.
+    // Client: Sends connection attributes in Protocol::HandshakeResponse41.
+    int CLIENT_CONNECT_ATTRS                  = 0x00100000;
+
+    // Server:Understands length-encoded integer for auth response data in
+    // Protocol::HandshakeResponse41.
+    // Client:Length of auth response data in Protocol::HandshakeResponse41 is a
+    // length-encoded integer.
+    // Note: The flag was introduced in 5.6.6, but had the wrong value.
+    int CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA = 0x00200000;
+
+    // Server: Announces support for expired password extension.
+    // Client: Can handle expired passwords.
+    int CLIENT_CAN_HANDLE_EXPIRED_PASSWORDS   = 0x00400000;
+
+    // Server: Can set SERVER_SESSION_STATE_CHANGED in the Status Flags and send
+    // session-state change data after a OK packet.
+    // Client: Expects the server to send sesson-state changes after a OK
+    // packet.
+    int CLIENT_SESSION_TRACK                  = 0x00800000;
+
+    /**
+     * Server: Can send OK after a Text Resultset. Client: Expects an OK
+     * (instead of EOF) after the resultset rows of a Text Resultset.
+     * Background:To support CLIENT_SESSION_TRACK, additional information must
+     * be sent after all successful commands. Although the OK packet is
+     * extensible, the EOF packet is not due to the overlap of its bytes with
+     * the content of the Text Resultset Row. Therefore, the EOF packet in the
+     * Text Resultset is replaced with an OK packet. EOF packets are deprecated
+     * as of MySQL 5.7.5.
+     */
+    int CLIENT_DEPRECATE_EOF                  = 0x01000000;
+
+}

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/AuthSwitchResponsePacket.java

@@ -0,0 +1,21 @@
+package com.alibaba.otter.canal.parse.driver.mysql.packets.client;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+
+import com.alibaba.otter.canal.parse.driver.mysql.packets.CommandPacket;
+
+public class AuthSwitchResponsePacket extends CommandPacket {
+
+    public byte[] authData;
+
+    public void fromBytes(byte[] data) {
+    }
+
+    public byte[] toBytes() throws IOException {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        out.write(authData);
+        return out.toByteArray();
+    }
+
+}

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/ClientAuthenticationPacket.java

@@ -6,6 +6,7 @@ import java.security.NoSuchAlgorithmException;
 
 import org.apache.commons.lang.StringUtils;
 
+import com.alibaba.otter.canal.parse.driver.mysql.packets.Capability;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.PacketWithHeaderPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
 import com.alibaba.otter.canal.parse.driver.mysql.utils.MSC;
@@ -13,12 +14,17 @@ import com.alibaba.otter.canal.parse.driver.mysql.utils.MySQLPasswordEncrypter;
 
 public class ClientAuthenticationPacket extends PacketWithHeaderPacket {
 
+    private int    clientCapability = Capability.CLIENT_LONG_PASSWORD | Capability.CLIENT_LONG_FLAG
+                                      | Capability.CLIENT_PROTOCOL_41 | Capability.CLIENT_INTERACTIVE
+                                      | Capability.CLIENT_TRANSACTIONS | Capability.CLIENT_SECURE_CONNECTION
+                                      | Capability.CLIENT_MULTI_STATEMENTS | Capability.CLIENT_PLUGIN_AUTH;
     private String username;
     private String password;
     private byte   charsetNumber;
     private String databaseName;
     private int    serverCapabilities;
     private byte[] scrumbleBuff;
+    private byte[] authPluginName;
 
     public void fromBytes(byte[] data) {
         // bypass since nowhere to use.
@@ -36,6 +42,7 @@ public class ClientAuthenticationPacket extends PacketWithHeaderPacket {
      *  n (Null-Terminated String)   user
      *  n (Length Coded Binary)      scramble_buff (1 + x bytes)
      *  n (Null-Terminated String)   databasename (optional)
+     *  n (Null-Terminated String)   auth plugin name (optional)
      * </pre>
      * 
      * @throws IOException
@@ -43,13 +50,7 @@ public class ClientAuthenticationPacket extends PacketWithHeaderPacket {
     public byte[] toBytes() throws IOException {
         ByteArrayOutputStream out = new ByteArrayOutputStream();
         // 1. write client_flags
-        // 1|4|512|1024|8192|32768
+        ByteHelper.writeUnsignedIntLittleEndian(clientCapability, out); // remove
-        /**
-         * CLIENT_LONG_PASSWORD CLIENT_LONG_FLAG CLIENT_PROTOCOL_41
-         * CLIENT_INTERACTIVE CLIENT_TRANSACTIONS CLIENT_SECURE_CONNECTION
-         * CLIENT_MULTI_STATEMENTS;
-         */
-        ByteHelper.writeUnsignedIntLittleEndian(1 | 4 | 512 | 8192 | 32768 | 0x00010000, out); // remove
         // client_interactive
         // feature
 
@@ -76,6 +77,10 @@ public class ClientAuthenticationPacket extends PacketWithHeaderPacket {
         if (getDatabaseName() != null) {
             ByteHelper.writeNullTerminatedString(getDatabaseName(), out);
         }
+        // 8 . (Null-Terminated String) auth plugin name (optional)
+        if (getAuthPluginName() != null) {
+            ByteHelper.writeNullTerminated(getAuthPluginName(), out);
+        }
         // end write
         return out.toByteArray();
     }
@@ -106,6 +111,9 @@ public class ClientAuthenticationPacket extends PacketWithHeaderPacket {
 
     public void setDatabaseName(String databaseName) {
         this.databaseName = databaseName;
+        if (databaseName != null) {
+            this.clientCapability |= Capability.CLIENT_CONNECT_WITH_DB;
+        }
     }
 
     public String getDatabaseName() {
@@ -128,4 +136,15 @@ public class ClientAuthenticationPacket extends PacketWithHeaderPacket {
         return scrumbleBuff;
     }
 
+    public byte[] getAuthPluginName() {
+        return authPluginName;
+    }
+
+    public void setAuthPluginName(byte[] authPluginName) {
+        this.authPluginName = authPluginName;
+        if (authPluginName != null) {
+            this.clientCapability |= Capability.CLIENT_PLUGIN_AUTH;
+        }
+    }
+
 }

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/server/AuthSwitchRequestMoreData.java

@@ -0,0 +1,25 @@
+package com.alibaba.otter.canal.parse.driver.mysql.packets.server;
+
+import java.io.IOException;
+
+import com.alibaba.otter.canal.parse.driver.mysql.packets.CommandPacket;
+import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
+
+public class AuthSwitchRequestMoreData extends CommandPacket {
+
+    public int    status;
+    public byte[] authData;
+
+    public void fromBytes(byte[] data) {
+        int index = 0;
+        // 1. read status
+        status = data[index];
+        index += 1;
+        authData = ByteHelper.readNullTerminatedBytes(data, index);
+    }
+
+    public byte[] toBytes() throws IOException {
+        return null;
+    }
+
+}

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/server/AuthSwitchRequestPacket.java

@@ -0,0 +1,29 @@
+package com.alibaba.otter.canal.parse.driver.mysql.packets.server;
+
+import java.io.IOException;
+
+import com.alibaba.otter.canal.parse.driver.mysql.packets.CommandPacket;
+import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
+
+public class AuthSwitchRequestPacket extends CommandPacket {
+
+    public int    status;
+    public String authName;
+    public byte[] authData;
+
+    public void fromBytes(byte[] data) {
+        int index = 0;
+        // 1. read status
+        status = data[index];
+        index += 1;
+        byte[] authName = ByteHelper.readNullTerminatedBytes(data, index);
+        this.authName = new String(authName);
+        index += authName.length + 1;
+        authData = ByteHelper.readNullTerminatedBytes(data, index);
+    }
+
+    public byte[] toBytes() throws IOException {
+        return null;
+    }
+
+}

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/server/HandshakeInitializationPacket.java

@@ -2,6 +2,7 @@ package com.alibaba.otter.canal.parse.driver.mysql.packets.server;
 
 import java.io.IOException;
 
+import com.alibaba.otter.canal.parse.driver.mysql.packets.Capability;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.PacketWithHeaderPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
@@ -23,6 +24,7 @@ public class HandshakeInitializationPacket extends PacketWithHeaderPacket {
     public byte   serverCharsetNumber;
     public int    serverStatus;
     public byte[] restOfScrambleBuff;
+    public byte[] authPluginName;
 
     public HandshakeInitializationPacket(){
     }
@@ -66,6 +68,7 @@ public class HandshakeInitializationPacket extends PacketWithHeaderPacket {
         // 5. read server_capabilities
         this.serverCapabilities = ByteHelper.readUnsignedShortLittleEndian(data, index);
         index += 2;
+        if (data.length > index) {
             // 6. read server_language
             this.serverCharsetNumber = data[index];
             index++;
@@ -73,14 +76,33 @@ public class HandshakeInitializationPacket extends PacketWithHeaderPacket {
             this.serverStatus = ByteHelper.readUnsignedShortLittleEndian(data, index);
             index += 2;
             // 8. bypass filtered bytes
-        index += 13;
+            int capabilityFlags2 = ByteHelper.readUnsignedShortLittleEndian(data, index);
+            index += 2;
+            int capabilities = (capabilityFlags2 << 16) | this.serverCapabilities;
+            // int authPluginDataLen = -1;
+            // if ((capabilities & Capability.CLIENT_PLUGIN_AUTH) != 0) {
+            // authPluginDataLen = data[index];
+            // }
+            index += 1;
+            index += 10;
             // 9. read rest of scramble_buff
-        this.restOfScrambleBuff = ByteHelper.readFixedLengthBytes(data, index, 12); // 虽然Handshake
+            if ((capabilities & Capability.CLIENT_SECURE_CONNECTION) != 0) {
-                                                                                    // Initialization
+                // int len = Math.max(13, authPluginDataLen - 8);
+                // this.authPluginDataPart2 =
+                // buffer.readFixedLengthString(len);// scramble2
+
                 // Packet规定最后13个byte是剩下的scrumble,
                 // 但实际上最后一个字节是0, 不应该包含在scrumble中.
+                this.restOfScrambleBuff = ByteHelper.readFixedLengthBytes(data, index, 12);
+            }
+
+            index += 12 + 1;
+            if ((capabilities & Capability.CLIENT_PLUGIN_AUTH) != 0) {
+                this.authPluginName = ByteHelper.readNullTerminatedBytes(data, index);
+            }
             // end read
         }
+    }
 
     /**
      * Bypass implementing it, 'cause nowhere to use it.

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/MySQLPasswordEncrypter.java

@@ -1,10 +1,47 @@
 package com.alibaba.otter.canal.parse.driver.mysql.utils;
 
+import java.security.DigestException;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 
 public class MySQLPasswordEncrypter {
 
+    private static final int CACHING_SHA2_DIGEST_LENGTH = 32;
+
+    public static byte[] scrambleCachingSha2(byte[] password, byte[] seed) throws DigestException {
+        MessageDigest md;
+        try {
+            md = MessageDigest.getInstance("SHA-256");
+        } catch (NoSuchAlgorithmException ex) {
+            throw new DigestException(ex);
+        }
+
+        byte[] dig1 = new byte[CACHING_SHA2_DIGEST_LENGTH];
+        byte[] dig2 = new byte[CACHING_SHA2_DIGEST_LENGTH];
+        byte[] scramble1 = new byte[CACHING_SHA2_DIGEST_LENGTH];
+
+        // SHA2(src) => digest_stage1
+        md.update(password, 0, password.length);
+        md.digest(dig1, 0, CACHING_SHA2_DIGEST_LENGTH);
+        md.reset();
+
+        // SHA2(digest_stage1) => digest_stage2
+        md.update(dig1, 0, dig1.length);
+        md.digest(dig2, 0, CACHING_SHA2_DIGEST_LENGTH);
+        md.reset();
+
+        // SHA2(digest_stage2, m_rnd) => scramble_stage1
+        md.update(dig2, 0, dig1.length);
+        md.update(seed, 0, seed.length);
+        md.digest(scramble1, 0, CACHING_SHA2_DIGEST_LENGTH);
+
+        // XOR(digest_stage1, scramble_stage1) => scramble
+        byte[] mysqlScrambleBuff = new byte[CACHING_SHA2_DIGEST_LENGTH];
+        xorString(dig1, mysqlScrambleBuff, scramble1, CACHING_SHA2_DIGEST_LENGTH);
+
+        return mysqlScrambleBuff;
+    }
+
     public static final byte[] scramble411(byte[] pass, byte[] seed) throws NoSuchAlgorithmException {
         MessageDigest md = MessageDigest.getInstance("SHA-1");
         byte[] pass1 = md.digest(pass);
@@ -71,4 +108,13 @@ public class MySQLPasswordEncrypter {
         return result;
     }
 
+    private static void xorString(byte[] from, byte[] to, byte[] scramble, int length) {
+        int pos = 0;
+        int scrambleLength = scramble.length;
+        while (pos < length) {
+            to[pos] = (byte) (from[pos] ^ scramble[pos % scrambleLength]);
+            pos++;
+        }
+    }
+
 }

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/PacketManager.java

@@ -1,5 +1,6 @@
 package com.alibaba.otter.canal.parse.driver.mysql.utils;
 
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 
 import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket;
@@ -28,7 +29,11 @@ public abstract class PacketManager {
     }
 
     public static void writePkg(SocketChannel ch, byte[]... srcs) throws IOException {
-        ch.write(srcs);
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        for (byte[] src : srcs) {
+            out.write(src);
+        }
+        ch.write(out.toByteArray());
     }
 
     public static void writeBody(SocketChannel ch, byte[] body) throws IOException {

driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnectorTest.java

@@ -4,10 +4,11 @@ import java.io.IOException;
 import java.net.InetSocketAddress;
 
 import org.junit.Assert;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket;
-
+@Ignore
 public class MysqlConnectorTest {
 
     @Test

example/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.1.3-SNAPSHOT</version>
+		<version>1.1.4-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

example/src/main/java/com/alibaba/otter/canal/example/rocketmq/AbstractRocektMQTest.java

@@ -7,5 +7,9 @@ public abstract class AbstractRocektMQTest extends BaseCanalClientTest {
     public static String topic       = "example";
     public static String groupId     = "group";
     public static String nameServers = "localhost:9876";
-
+    public static String accessKey   = "";
+    public static String secretKey   = "";
+    public static boolean enableMessageTrace = false;
+    public static String accessChannel = "local";
+    public static String namespace = "";
 }

example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientExample.java

@@ -1,16 +1,13 @@
 package com.alibaba.otter.canal.example.rocketmq;
 
+import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector;
+import com.alibaba.otter.canal.protocol.Message;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.util.Assert;
 
-import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector;
-import com.alibaba.otter.canal.example.kafka.AbstractKafkaTest;
-import com.alibaba.otter.canal.protocol.Message;
-
 /**
  * Kafka client example
  *
@@ -38,12 +35,24 @@ public class CanalRocketMQClientExample extends AbstractRocektMQTest {
         connector = new RocketMQCanalConnector(nameServers, topic, groupId, 500, false);
     }
 
+    public CanalRocketMQClientExample(String nameServers, String topic, String groupId, boolean enableMessageTrace,
+        String accessKey, String secretKey, String accessChannel, String namespace) {
+        connector = new RocketMQCanalConnector(nameServers, topic, groupId, accessKey,
+            secretKey, -1, false, enableMessageTrace,
+            null, accessChannel, namespace);
+    }
+
     public static void main(String[] args) {
         try {
             final CanalRocketMQClientExample rocketMQClientExample = new CanalRocketMQClientExample(nameServers,
                 topic,
-                groupId);
+                groupId,
-            logger.info("## Start the rocketmq consumer: {}-{}", AbstractKafkaTest.topic, AbstractKafkaTest.groupId);
+                enableMessageTrace,
+                accessKey,
+                secretKey,
+                accessChannel,
+                namespace);
+            logger.info("## Start the rocketmq consumer: {}-{}", topic, groupId);
             rocketMQClientExample.start();
             logger.info("## The canal rocketmq consumer is running now ......");
             Runtime.getRuntime().addShutdownHook(new Thread() {
@@ -108,7 +117,7 @@ public class CanalRocketMQClientExample extends AbstractRocektMQTest {
                 connector.connect();
                 connector.subscribe();
                 while (running) {
-                    List<Message> messages = connector.getListWithoutAck(100L, TimeUnit.MILLISECONDS); // 获取message
+                    List<Message> messages = connector.getListWithoutAck(1000L, TimeUnit.MILLISECONDS); // 获取message
                     for (Message message : messages) {
                         long batchId = message.getId();
                         int size = message.getEntries().size();

example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientFlatMessageExample.java

@@ -43,7 +43,7 @@ public class CanalRocketMQClientFlatMessageExample extends AbstractRocektMQTest
             final CanalRocketMQClientFlatMessageExample rocketMQClientExample = new CanalRocketMQClientFlatMessageExample(nameServers,
                 topic,
                 groupId);
-            logger.info("## Start the rocketmq consumer: {}-{}", AbstractKafkaTest.topic, AbstractKafkaTest.groupId);
+            logger.info("## Start the rocketmq consumer: {}-{}", topic, groupId);
             rocketMQClientExample.start();
             logger.info("## The canal rocketmq consumer is running now ......");
             Runtime.getRuntime().addShutdownHook(new Thread() {

filter/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.1.3-SNAPSHOT</version>
+		<version>1.1.4-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

instance/core/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.1.3-SNAPSHOT</version>
+		<version>1.1.4-SNAPSHOT</version>
 		<relativePath>../../pom.xml</relativePath>
 	</parent>
 	<artifactId>canal.instance.core</artifactId>

instance/core/src/main/java/com/alibaba/otter/canal/instance/core/AbstractCanalInstance.java

@@ -53,11 +53,15 @@ public class AbstractCanalInstance extends AbstractCanalLifeCycle implements Can
                 // 处理group的模式
                 List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
                 for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
+                    if(singleEventParser instanceof AbstractEventParser) {
                         ((AbstractEventParser) singleEventParser).setEventFilter(aviaterFilter);
                     }
+                }
             } else {
+                if(eventParser instanceof AbstractEventParser) {
                     ((AbstractEventParser) eventParser).setEventFilter(aviaterFilter);
                 }
+            }
 
         }