[maven-release-plugin] prepare release canal-1.1.3

flatMessage模式下默认canal.mq.lingerMs=100

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.3</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.3</version>
         </dependency>
         <dependency>
             <groupId>joda-time</groupId>

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.3</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

@@ -17,6 +17,7 @@ import org.elasticsearch.client.transport.TransportClient;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.transport.client.PreBuiltTransportClient;
+import org.slf4j.MDC;
 
 import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.otter.canal.client.adapter.OuterAdapter;
@@ -29,8 +30,12 @@ import com.alibaba.otter.canal.client.adapter.es.monitor.ESConfigMonitor;
 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 com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
-import org.slf4j.MDC;
+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;
+
 
 /**
  * ES外部适配器
@@ -71,7 +76,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 +145,7 @@ public class ESAdapter implements OuterAdapter {
         }
     }
 
+    @Override
     public void sync(List<Dml> dmls) {
         if (dmls == null || dmls.isEmpty()) {
             return;
@@ -151,6 +156,7 @@ public class ESAdapter implements OuterAdapter {
             }
         }
         esSyncService.commit(); // 批次统一提交
+
     }
 
     private void sync(Dml dml) {
@@ -241,7 +247,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/SqlParser.java

@@ -9,8 +9,16 @@ 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.expr.SQLBinaryOpExpr;
-import com.alibaba.fastsql.sql.ast.statement.*;
+import com.alibaba.fastsql.sql.ast.expr.SQLCaseExpr;
+import com.alibaba.fastsql.sql.ast.expr.SQLIdentifierExpr;
+import com.alibaba.fastsql.sql.ast.expr.SQLMethodInvokeExpr;
+import com.alibaba.fastsql.sql.ast.expr.SQLPropertyExpr;
+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;

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

@@ -1,7 +1,11 @@
 package com.alibaba.otter.canal.client.adapter.es.service;
 
 import java.sql.SQLException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+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.AtomicLong;

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

@@ -7,7 +7,6 @@ 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 +22,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

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;

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

@@ -1,6 +1,10 @@
 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;
 

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

@@ -1,6 +1,10 @@
 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;
 

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

@@ -1,6 +1,10 @@
 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;
 

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

@@ -1,6 +1,10 @@
 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;
 

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

@@ -1,6 +1,10 @@
 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;
 

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

@@ -1,6 +1,10 @@
 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;

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.3</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.alibaba.otter</groupId>

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

@@ -26,7 +26,6 @@ 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 org.slf4j.MDC;
 
 /**
  * HBase外部适配器
@@ -60,7 +59,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 +102,7 @@ public class HbaseAdapter implements OuterAdapter {
         }
     }
 
+    @Override
     public void sync(List<Dml> dmls) {
         if (dmls == null || dmls.isEmpty()) {
             return;
@@ -113,7 +112,7 @@ public class HbaseAdapter implements OuterAdapter {
         }
     }
 
-    public void sync(Dml dml) {
+    private void sync(Dml dml) {
         if (dml == null) {
             return;
         }
@@ -224,7 +223,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/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/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.3</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.3</version>
         </dependency>
         <dependency>
             <groupId>org.yaml</groupId>

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

@@ -40,11 +40,11 @@ if [ -z "$JAVA" ]; then
   fi
 fi
 
-case "$#" 
+case "$#"
 in
-0 ) 
+0 )
   ;;
-2 ) 
+2 )
   if [ "$1" = "debug" ]; then
     DEBUG_PORT=$2
     DEBUG_SUSPEND="n"
@@ -76,8 +76,8 @@ 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"
-cd $current_path
+cd $current_path

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;

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/logback.xml

@@ -8,17 +8,17 @@
 	</appender>
 
 	<appender name="CANAL-ROOT" class="ch.qos.logback.classic.sift.SiftingAppender">
-		<discriminator>
+        <discriminator>
-			<Key>adapter</Key>
+            <Key>adapter</Key>
-			<DefaultValue>adapter</DefaultValue>
+            <DefaultValue>adapter</DefaultValue>
-		</discriminator>
+        </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.3</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.3</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.3</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

@@ -14,7 +14,6 @@ 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 +25,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 +70,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 +136,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);
@@ -304,6 +297,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/service/RdbEtlService.java

@@ -1,6 +1,10 @@
 package com.alibaba.otter.canal.client.adapter.rdb.service;
 
-import java.sql.*;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;

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/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java

@@ -1,6 +1,11 @@
 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.Test;

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.3</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

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

@@ -1,325 +1,326 @@
-package com.alibaba.otter.canal.client.kafka;
+package com.alibaba.otter.canal.client.kafka;
-
+
-import java.util.*;
+import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeUnit;
-
+
-import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
-
+
-import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSON;
-import com.alibaba.otter.canal.client.CanalMQConnector;
+import com.alibaba.otter.canal.client.CanalMQConnector;
-import com.alibaba.otter.canal.client.impl.SimpleCanalConnector;
+import com.alibaba.otter.canal.client.impl.SimpleCanalConnector;
-import com.alibaba.otter.canal.protocol.FlatMessage;
+import com.alibaba.otter.canal.protocol.FlatMessage;
-import com.alibaba.otter.canal.protocol.Message;
+import com.alibaba.otter.canal.protocol.Message;
-import com.alibaba.otter.canal.protocol.exception.CanalClientException;
+import com.alibaba.otter.canal.protocol.exception.CanalClientException;
-import com.google.common.collect.Lists;
+import com.google.common.collect.Lists;
-
+
-/**
+/**
- * canal kafka 数据操作客户端
+ * canal kafka 数据操作客户端
- *
+ *
- * <pre>
+ * <pre>
- * 注意点：
+ * 注意点：
- * 1. 相比于canal {@linkplain SimpleCanalConnector}, 这里get和ack操作不能有并发, 必须是一个线程执行get后，内存里执行完毕ack后再取下一个get
+ * 1. 相比于canal {@linkplain SimpleCanalConnector}, 这里get和ack操作不能有并发, 必须是一个线程执行get后，内存里执行完毕ack后再取下一个get
- * </pre>
+ * </pre>
- *
+ *
- * @author machengyuan @ 2018-6-12
+ * @author machengyuan @ 2018-6-12
- * @version 1.1.1
+ * @version 1.1.1
- */
+ */
-public class KafkaCanalConnector implements CanalMQConnector {
+public class KafkaCanalConnector implements CanalMQConnector {
-
+
-    protected KafkaConsumer<String, Message> kafkaConsumer;
+    protected KafkaConsumer<String, Message> kafkaConsumer;
-    protected KafkaConsumer<String, String>  kafkaConsumer2;                            // 用于扁平message的数据消费
+    protected KafkaConsumer<String, String>  kafkaConsumer2;                            // 用于扁平message的数据消费
-    protected String                         topic;
+    protected String                         topic;
-    protected Integer                        partition;
+    protected Integer                        partition;
-    protected Properties                     properties;
+    protected Properties                     properties;
-    protected volatile boolean               connected      = false;
+    protected volatile boolean               connected      = false;
-    protected volatile boolean               running        = false;
+    protected volatile boolean               running        = false;
-    protected boolean                        flatMessage;
+    protected boolean                        flatMessage;
-
+
-    private Map<Integer, Long>               currentOffsets = new ConcurrentHashMap<>();
+    private Map<Integer, Long>               currentOffsets = new ConcurrentHashMap<>();
-
+
-    public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId, Integer batchSize,
+    public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId, Integer batchSize,
-                               boolean flatMessage){
+                               boolean flatMessage){
-        this.topic = topic;
+        this.topic = topic;
-        this.partition = partition;
+        this.partition = partition;
-        this.flatMessage = flatMessage;
+        this.flatMessage = flatMessage;
-
+
-        properties = new Properties();
+        properties = new Properties();
-        properties.put("bootstrap.servers", servers);
+        properties.put("bootstrap.servers", servers);
-        properties.put("group.id", groupId);
+        properties.put("group.id", groupId);
-        properties.put("enable.auto.commit", false);
+        properties.put("enable.auto.commit", false);
-        properties.put("auto.commit.interval.ms", "1000");
+        properties.put("auto.commit.interval.ms", "1000");
-        properties.put("auto.offset.reset", "latest"); // 如果没有offset则从最后的offset开始读
+        properties.put("auto.offset.reset", "latest"); // 如果没有offset则从最后的offset开始读
-        properties.put("request.timeout.ms", "40000"); // 必须大于session.timeout.ms的设置
+        properties.put("request.timeout.ms", "40000"); // 必须大于session.timeout.ms的设置
-        properties.put("session.timeout.ms", "30000"); // 默认为30秒
+        properties.put("session.timeout.ms", "30000"); // 默认为30秒
-        if (batchSize == null) {
+        properties.put("isolation.level", "read_committed");
-            batchSize = 100;
+        if (batchSize == null) {
-        }
+            batchSize = 100;
-        properties.put("max.poll.records", batchSize.toString());
+        }
-        properties.put("key.deserializer", StringDeserializer.class.getName());
+        properties.put("max.poll.records", batchSize.toString());
-        if (!flatMessage) {
+        properties.put("key.deserializer", StringDeserializer.class.getName());
-            properties.put("value.deserializer", MessageDeserializer.class.getName());
+        if (!flatMessage) {
-        } else {
+            properties.put("value.deserializer", MessageDeserializer.class.getName());
-            properties.put("value.deserializer", StringDeserializer.class.getName());
+        } else {
-        }
+            properties.put("value.deserializer", StringDeserializer.class.getName());
-    }
+        }
-
+    }
-    /**
+
-     * 打开连接
+    /**
-     */
+     * 打开连接
-    @Override
+     */
-    public void connect() {
+    @Override
-        if (connected) {
+    public void connect() {
-            return;
+        if (connected) {
-        }
+            return;
-
+        }
-        connected = true;
+
-        if (kafkaConsumer == null && !flatMessage) {
+        connected = true;
-            kafkaConsumer = new KafkaConsumer<String, Message>(properties);
+        if (kafkaConsumer == null && !flatMessage) {
-
+            kafkaConsumer = new KafkaConsumer<String, Message>(properties);
-        }
+
-        if (kafkaConsumer2 == null && flatMessage) {
+        }
-            kafkaConsumer2 = new KafkaConsumer<String, String>(properties);
+        if (kafkaConsumer2 == null && flatMessage) {
-        }
+            kafkaConsumer2 = new KafkaConsumer<String, String>(properties);
-    }
+        }
-
+    }
-    /**
+
-     * 关闭链接
+    /**
-     */
+     * 关闭链接
-    @Override
+     */
-    public void disconnect() {
+    @Override
-        if (kafkaConsumer != null) {
+    public void disconnect() {
-            kafkaConsumer.close();
+        if (kafkaConsumer != null) {
-            kafkaConsumer = null;
+            kafkaConsumer.close();
-        }
+            kafkaConsumer = null;
-        if (kafkaConsumer2 != null) {
+        }
-            kafkaConsumer2.close();
+        if (kafkaConsumer2 != null) {
-            kafkaConsumer2 = null;
+            kafkaConsumer2.close();
-        }
+            kafkaConsumer2 = null;
-
+        }
-        connected = false;
+
-    }
+        connected = false;
-
+    }
-    protected void waitClientRunning() {
+
-        running = true;
+    protected void waitClientRunning() {
-    }
+        running = true;
-
+    }
-    @Override
+
-    public boolean checkValid() {
+    @Override
-        return true;// 默认都放过
+    public boolean checkValid() {
-    }
+        return true;// 默认都放过
-
+    }
-    /**
+
-     * 订阅topic
+    /**
-     */
+     * 订阅topic
-    @Override
+     */
-    public void subscribe() {
+    @Override
-        waitClientRunning();
+    public void subscribe() {
-        if (!running) {
+        waitClientRunning();
-            return;
+        if (!running) {
-        }
+            return;
-
+        }
-        if (partition == null) {
+
-            if (kafkaConsumer != null) {
+        if (partition == null) {
-                kafkaConsumer.subscribe(Collections.singletonList(topic));
+            if (kafkaConsumer != null) {
-            }
+                kafkaConsumer.subscribe(Collections.singletonList(topic));
-            if (kafkaConsumer2 != null) {
+            }
-                kafkaConsumer2.subscribe(Collections.singletonList(topic));
+            if (kafkaConsumer2 != null) {
-            }
+                kafkaConsumer2.subscribe(Collections.singletonList(topic));
-        } else {
+            }
-            TopicPartition topicPartition = new TopicPartition(topic, partition);
+        } else {
-            if (kafkaConsumer != null) {
+            TopicPartition topicPartition = new TopicPartition(topic, partition);
-                kafkaConsumer.assign(Collections.singletonList(topicPartition));
+            if (kafkaConsumer != null) {
-            }
+                kafkaConsumer.assign(Collections.singletonList(topicPartition));
-            if (kafkaConsumer2 != null) {
+            }
-                kafkaConsumer2.assign(Collections.singletonList(topicPartition));
+            if (kafkaConsumer2 != null) {
-            }
+                kafkaConsumer2.assign(Collections.singletonList(topicPartition));
-        }
+            }
-    }
+        }
-
+    }
-    /**
+
-     * 取消订阅
+    /**
-     */
+     * 取消订阅
-    @Override
+     */
-    public void unsubscribe() {
+    @Override
-        waitClientRunning();
+    public void unsubscribe() {
-        if (!running) {
+        waitClientRunning();
-            return;
+        if (!running) {
-        }
+            return;
-
+        }
-        if (kafkaConsumer != null) {
+
-            kafkaConsumer.unsubscribe();
+        if (kafkaConsumer != null) {
-        }
+            kafkaConsumer.unsubscribe();
-        if (kafkaConsumer2 != null) {
+        }
-            kafkaConsumer2.unsubscribe();
+        if (kafkaConsumer2 != null) {
-        }
+            kafkaConsumer2.unsubscribe();
-    }
+        }
-
+    }
-    @Override
+
-    public List<Message> getList(Long timeout, TimeUnit unit) throws CanalClientException {
+    @Override
-        waitClientRunning();
+    public List<Message> getList(Long timeout, TimeUnit unit) throws CanalClientException {
-        if (!running) {
+        waitClientRunning();
-            return Lists.newArrayList();
+        if (!running) {
-        }
+            return Lists.newArrayList();
-
+        }
-        List<Message> messages = getListWithoutAck(timeout, unit);
+
-        if (messages != null && !messages.isEmpty()) {
+        List<Message> messages = getListWithoutAck(timeout, unit);
-            this.ack();
+        if (messages != null && !messages.isEmpty()) {
-        }
+            this.ack();
-        return messages;
+        }
-    }
+        return messages;
-
+    }
-    @Override
+
-    public List<Message> getListWithoutAck(Long timeout, TimeUnit unit) throws CanalClientException {
+    @Override
-        waitClientRunning();
+    public List<Message> getListWithoutAck(Long timeout, TimeUnit unit) throws CanalClientException {
-        if (!running) {
+        waitClientRunning();
-            return Lists.newArrayList();
+        if (!running) {
-        }
+            return Lists.newArrayList();
-
+        }
-        ConsumerRecords<String, Message> records = kafkaConsumer.poll(unit.toMillis(timeout));
+
-
+        ConsumerRecords<String, Message> records = kafkaConsumer.poll(unit.toMillis(timeout));
-        currentOffsets.clear();
+
-        for (TopicPartition topicPartition : records.partitions()) {
+        currentOffsets.clear();
-            currentOffsets.put(topicPartition.partition(), kafkaConsumer.position(topicPartition));
+        for (TopicPartition topicPartition : records.partitions()) {
-        }
+            currentOffsets.put(topicPartition.partition(), kafkaConsumer.position(topicPartition));
-
+        }
-        if (!records.isEmpty()) {
+
-            List<Message> messages = new ArrayList<>();
+        if (!records.isEmpty()) {
-            for (ConsumerRecord<String, Message> record : records) {
+            List<Message> messages = new ArrayList<>();
-                messages.add(record.value());
+            for (ConsumerRecord<String, Message> record : records) {
-            }
+                messages.add(record.value());
-            return messages;
+            }
-        }
+            return messages;
-        return Lists.newArrayList();
+        }
-    }
+        return Lists.newArrayList();
-
+    }
-    @Override
+
-    public List<FlatMessage> getFlatList(Long timeout, TimeUnit unit) throws CanalClientException {
+    @Override
-        waitClientRunning();
+    public List<FlatMessage> getFlatList(Long timeout, TimeUnit unit) throws CanalClientException {
-        if (!running) {
+        waitClientRunning();
-            return Lists.newArrayList();
+        if (!running) {
-        }
+            return Lists.newArrayList();
-
+        }
-        List<FlatMessage> messages = getFlatListWithoutAck(timeout, unit);
+
-        if (messages != null && !messages.isEmpty()) {
+        List<FlatMessage> messages = getFlatListWithoutAck(timeout, unit);
-            this.ack();
+        if (messages != null && !messages.isEmpty()) {
-        }
+            this.ack();
-        return messages;
+        }
-    }
+        return messages;
-
+    }
-    @Override
+
-    public List<FlatMessage> getFlatListWithoutAck(Long timeout, TimeUnit unit) throws CanalClientException {
+    @Override
-        waitClientRunning();
+    public List<FlatMessage> getFlatListWithoutAck(Long timeout, TimeUnit unit) throws CanalClientException {
-        if (!running) {
+        waitClientRunning();
-            return Lists.newArrayList();
+        if (!running) {
-        }
+            return Lists.newArrayList();
-
+        }
-        ConsumerRecords<String, String> records = kafkaConsumer2.poll(unit.toMillis(timeout));
+
-
+        ConsumerRecords<String, String> records = kafkaConsumer2.poll(unit.toMillis(timeout));
-        currentOffsets.clear();
+
-        for (TopicPartition topicPartition : records.partitions()) {
+        currentOffsets.clear();
-            currentOffsets.put(topicPartition.partition(), kafkaConsumer2.position(topicPartition));
+        for (TopicPartition topicPartition : records.partitions()) {
-        }
+            currentOffsets.put(topicPartition.partition(), kafkaConsumer2.position(topicPartition));
-
+        }
-        if (!records.isEmpty()) {
+
-            List<FlatMessage> flatMessages = new ArrayList<>();
+        if (!records.isEmpty()) {
-            for (ConsumerRecord<String, String> record : records) {
+            List<FlatMessage> flatMessages = new ArrayList<>();
-                String flatMessageJson = record.value();
+            for (ConsumerRecord<String, String> record : records) {
-                FlatMessage flatMessage = JSON.parseObject(flatMessageJson, FlatMessage.class);
+                String flatMessageJson = record.value();
-                flatMessages.add(flatMessage);
+                FlatMessage flatMessage = JSON.parseObject(flatMessageJson, FlatMessage.class);
-            }
+                flatMessages.add(flatMessage);
-
+            }
-            return flatMessages;
+
-        }
+            return flatMessages;
-        return Lists.newArrayList();
+        }
-    }
+        return Lists.newArrayList();
-
+    }
-    @Override
+
-    public void rollback() {
+    @Override
-        waitClientRunning();
+    public void rollback() {
-        if (!running) {
+        waitClientRunning();
-            return;
+        if (!running) {
-        }
+            return;
-        // 回滚所有分区
+        }
-        if (kafkaConsumer != null) {
+        // 回滚所有分区
-            for (Map.Entry<Integer, Long> entry : currentOffsets.entrySet()) {
+        if (kafkaConsumer != null) {
-                kafkaConsumer.seek(new TopicPartition(topic, entry.getKey()), entry.getValue() - 1);
+            for (Map.Entry<Integer, Long> entry : currentOffsets.entrySet()) {
-            }
+                kafkaConsumer.seek(new TopicPartition(topic, entry.getKey()), entry.getValue() - 1);
-        }
+            }
-        if (kafkaConsumer2 != null) {
+        }
-            for (Map.Entry<Integer, Long> entry : currentOffsets.entrySet()) {
+        if (kafkaConsumer2 != null) {
-                kafkaConsumer2.seek(new TopicPartition(topic, entry.getKey()), entry.getValue() - 1);
+            for (Map.Entry<Integer, Long> entry : currentOffsets.entrySet()) {
-            }
+                kafkaConsumer2.seek(new TopicPartition(topic, entry.getKey()), entry.getValue() - 1);
-        }
+            }
-    }
+        }
-
+    }
-    /**
+
-     * 提交offset，如果超过 session.timeout.ms 设置的时间没有ack则会抛出异常，ack失败
+    /**
-     */
+     * 提交offset，如果超过 session.timeout.ms 设置的时间没有ack则会抛出异常，ack失败
-    @Override
+     */
-    public void ack() {
+    @Override
-        waitClientRunning();
+    public void ack() {
-        if (!running) {
+        waitClientRunning();
-            return;
+        if (!running) {
-        }
+            return;
-
+        }
-        if (kafkaConsumer != null) {
+
-            kafkaConsumer.commitSync();
+        if (kafkaConsumer != null) {
-        }
+            kafkaConsumer.commitSync();
-        if (kafkaConsumer2 != null) {
+        }
-            kafkaConsumer2.commitSync();
+        if (kafkaConsumer2 != null) {
-        }
+            kafkaConsumer2.commitSync();
-    }
+        }
-
+    }
-    @Override
+
-    public void subscribe(String filter) throws CanalClientException {
+    @Override
-        throw new CanalClientException("mq not support this method");
+    public void subscribe(String filter) throws CanalClientException {
-    }
+        throw new CanalClientException("mq not support this method");
-
+    }
-    @Override
+
-    public Message get(int batchSize) throws CanalClientException {
+    @Override
-        throw new CanalClientException("mq not support this method");
+    public Message get(int batchSize) throws CanalClientException {
-    }
+        throw new CanalClientException("mq not support this method");
-
+    }
-    @Override
+
-    public Message get(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException {
+    @Override
-        throw new CanalClientException("mq not support this method");
+    public Message get(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException {
-    }
+        throw new CanalClientException("mq not support this method");
-
+    }
-    @Override
+
-    public Message getWithoutAck(int batchSize) throws CanalClientException {
+    @Override
-        throw new CanalClientException("mq not support this method");
+    public Message getWithoutAck(int batchSize) throws CanalClientException {
-    }
+        throw new CanalClientException("mq not support this method");
-
+    }
-    @Override
+
-    public Message getWithoutAck(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException {
+    @Override
-        throw new CanalClientException("mq not support this method");
+    public Message getWithoutAck(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException {
-    }
+        throw new CanalClientException("mq not support this method");
-
+    }
-    @Override
+
-    public void ack(long batchId) throws CanalClientException {
+    @Override
-        throw new CanalClientException("mq not support this method");
+    public void ack(long batchId) throws CanalClientException {
-    }
+        throw new CanalClientException("mq not support this method");
-
+    }
-    @Override
+
-    public void rollback(long batchId) throws CanalClientException {
+    @Override
-        throw new CanalClientException("mq not support this method");
+    public void rollback(long batchId) throws CanalClientException {
-    }
+        throw new CanalClientException("mq not support this method");
-
+    }
-    /**
+
-     * 重新设置sessionTime
+    /**
-     *
+     * 重新设置sessionTime
-     * @param timeout
+     *
-     * @param unit
+     * @param timeout
-     */
+     * @param unit
-    public void setSessionTimeout(Long timeout, TimeUnit unit) {
+     */
-        long t = unit.toMillis(timeout);
+    public void setSessionTimeout(Long timeout, TimeUnit unit) {
-        properties.put("request.timeout.ms", String.valueOf(t + 60000));
+        long t = unit.toMillis(timeout);
-        properties.put("session.timeout.ms", String.valueOf(t));
+        properties.put("request.timeout.ms", String.valueOf(t + 60000));
-    }
+        properties.put("session.timeout.ms", String.valueOf(t));
-
+    }
-}
+
+}

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.3</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<artifactId>canal.common</artifactId>

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.3</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

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.3</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

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)) {
@@ -68,13 +70,13 @@ public class CanalStater {
                     });
                     if (instanceDirs != null && instanceDirs.length > 0) {
                         List<String> instances = Lists.transform(Arrays.asList(instanceDirs),
-                                new Function<File, String>() {
+                            new Function<File, String>() {
 
-                                    @Override
+                                @Override
-                                    public String apply(File instanceDir) {
+                                public String apply(File instanceDir) {
-                                        return instanceDir.getName();
+                                    return instanceDir.getName();
-                                    }
+                                }
-                                });
+                            });
                         System.setProperty(CanalConstants.CANAL_DESTINATIONS, Joiner.on(",").join(instances));
                     }
                 }

deployer/src/main/resources/canal.properties

@@ -108,7 +108,7 @@ 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

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>
 

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.3</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,20 +68,40 @@ public class HandshakeInitializationPacket extends PacketWithHeaderPacket {
         // 5. read server_capabilities
         this.serverCapabilities = ByteHelper.readUnsignedShortLittleEndian(data, index);
         index += 2;
-        // 6. read server_language
+        if (data.length > index) {
-        this.serverCharsetNumber = data[index];
+            // 6. read server_language
-        index++;
+            this.serverCharsetNumber = data[index];
-        // 7. read server_status
+            index++;
-        this.serverStatus = ByteHelper.readUnsignedShortLittleEndian(data, index);
+            // 7. read server_status
-        index += 2;
+            this.serverStatus = ByteHelper.readUnsignedShortLittleEndian(data, index);
-        // 8. bypass filtered bytes
+            index += 2;
-        index += 13;
+            // 8. bypass filtered bytes
-        // 9. read rest of scramble_buff
+            int capabilityFlags2 = ByteHelper.readUnsignedShortLittleEndian(data, index);
-        this.restOfScrambleBuff = ByteHelper.readFixedLengthBytes(data, index, 12); // 虽然Handshake
+            index += 2;
-                                                                                    // Initialization
+            int capabilities = (capabilityFlags2 << 16) | this.serverCapabilities;
-        // Packet规定最后13个byte是剩下的scrumble,
+            // int authPluginDataLen = -1;
-        // 但实际上最后一个字节是0, 不应该包含在scrumble中.
+            // if ((capabilities & Capability.CLIENT_PLUGIN_AUTH) != 0) {
-        // end read
+            // authPluginDataLen = data[index];
+            // }
+            index += 1;
+            index += 10;
+            // 9. read rest of scramble_buff
+            if ((capabilities & Capability.CLIENT_SECURE_CONNECTION) != 0) {
+                // 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
+        }
     }
 
     /**

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 {

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.3</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

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.3</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.3</version>
 		<relativePath>../../pom.xml</relativePath>
 	</parent>
 	<artifactId>canal.instance.core</artifactId>

instance/manager/pom.xml

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

instance/pom.xml

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

instance/spring/pom.xml

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

instance/spring/src/test/resources/spring/default-instance.xml

@@ -140,7 +140,7 @@
 				<property name="address" value="${canal.instance.master.address}" />
 				<property name="username" value="${canal.instance.dbUsername:retl}" />
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 		<property name="standbyInfo">
@@ -148,7 +148,7 @@
 				<property name="address" value="${canal.instance.standby.address}" />
 				<property name="username" value="${canal.instance.dbUsername:retl}" />
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 		

instance/spring/src/test/resources/spring/file-instance.xml

@@ -128,7 +128,7 @@
 				<property name="address" value="${canal.instance.master.address}" />
 				<property name="username" value="${canal.instance.dbUsername:retl}" />
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 		<property name="standbyInfo">
@@ -136,7 +136,7 @@
 				<property name="address" value="${canal.instance.standby.address}" />
 				<property name="username" value="${canal.instance.dbUsername:retl}" />
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 		

instance/spring/src/test/resources/spring/group-instance.xml

@@ -123,7 +123,7 @@
 				<property name="address" value="${canal.instance.master1.address}" />
 				<property name="username" value="${canal.instance.dbUsername:retl}" />
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 		<property name="standbyInfo">
@@ -131,7 +131,7 @@
 				<property name="address" value="${canal.instance.standby1.address}" />
 				<property name="username" value="${canal.instance.dbUsername:retl}" />
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 		
@@ -203,7 +203,7 @@
 				<property name="address" value="${canal.instance.master2.address}" />
 				<property name="username" value="${canal.instance.dbUsername:retl}" />
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 		<property name="standbyInfo">
@@ -211,7 +211,7 @@
 				<property name="address" value="${canal.instance.standby2.address}" />
 				<property name="username" value="${canal.instance.dbUsername:retl}" />
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 		

instance/spring/src/test/resources/spring/memory-instance.xml

@@ -114,7 +114,7 @@
 				<property name="address" value="${canal.instance.master.address}" />
 				<property name="username" value="${canal.instance.dbUsername:retl}" />
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 		<property name="standbyInfo">
@@ -122,7 +122,7 @@
 				<property name="address" value="${canal.instance.standby.address}" />
 				<property name="username" value="${canal.instance.dbUsername:retl}" />
 				<property name="password" value="${canal.instance.dbPassword:retl}" />
-				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
 			</bean>
 		</property>
 		

meta/pom.xml

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

parse/pom.xml

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

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java

@@ -10,7 +10,6 @@ import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
-import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang.math.NumberUtils;
 import org.slf4j.Logger;
@@ -21,6 +20,7 @@ import com.alibaba.otter.canal.parse.driver.mysql.MysqlQueryExecutor;
 import com.alibaba.otter.canal.parse.driver.mysql.MysqlUpdateExecutor;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket;
+import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpCommandPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpGTIDCommandPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.client.RegisterSlaveCommandPacket;
@@ -131,9 +131,11 @@ public class MysqlConnection implements ErosaConnection {
         decoder.handle(LogEvent.QUERY_EVENT);
         decoder.handle(LogEvent.XID_EVENT);
         LogContext context = new LogContext();
-        // 若entry position存在gtid，则使用传入的gtid作为gtidSet 拼接的标准,否则同时开启gtid和tsdb时，会导致丢失gtid
+        // 若entry position存在gtid，则使用传入的gtid作为gtidSet
+        // 拼接的标准,否则同时开启gtid和tsdb时，会导致丢失gtid
         // 而当源端数据库gtid 有purged时会有如下类似报错
-        // 'errno = 1236, sqlstate = HY000 errmsg = The slave is connecting using CHANGE MASTER TO MASTER_AUTO_POSITION = 1 ...
+        // 'errno = 1236, sqlstate = HY000 errmsg = The slave is connecting
+        // using CHANGE MASTER TO MASTER_AUTO_POSITION = 1 ...
         if (StringUtils.isNotEmpty(gtid)) {
             decoder.handle(LogEvent.GTID_LOG_EVENT);
             context.setGtidSet(MysqlGTIDSet.parse(gtid));
@@ -417,7 +419,9 @@ public class MysqlConnection implements ErosaConnection {
             // '@@global.binlog_checksum'需要去掉单引号,在mysql 5.6.29下导致master退出
             update("set @master_binlog_checksum= @@global.binlog_checksum");
         } catch (Exception e) {
-            logger.warn("update master_binlog_checksum failed", e);
+            if (!StringUtils.contains(e.getMessage(), "Unknown system variable")) {
+                logger.warn("update master_binlog_checksum failed", e);
+            }
         }
 
         try {
@@ -434,7 +438,9 @@ public class MysqlConnection implements ErosaConnection {
             // mariadb针对特殊的类型，需要设置session变量
             update("SET @mariadb_slave_capability='" + LogEvent.MARIA_SLAVE_CAPABILITY_MINE + "'");
         } catch (Exception e) {
-            logger.warn("update mariadb_slave_capability failed", e);
+            if (!StringUtils.contains(e.getMessage(), "Unknown system variable")) {
+                logger.warn("update mariadb_slave_capability failed", e);
+            }
         }
 
         /**

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java

@@ -77,9 +77,7 @@ public class MemoryTableMeta implements TableMetaTSDB {
                 if (!StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "flush")
                     && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "grant")
                     && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "create user")
-                    && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "drop user")
+                    && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "drop user")) {
-                    && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "create database")
-                    && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "drop database")) {
                     repository.console(ddl);
                 }
             } catch (Throwable e) {

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/FastsqlSchemaTest.java

@@ -18,19 +18,12 @@ public class FastsqlSchemaTest {
     @Test
     public void testSimple() throws FileNotFoundException, IOException {
         SchemaRepository repository = new SchemaRepository(JdbcConstants.MYSQL);
-        String sql = "create table yushitai_test.card_record ( id bigint auto_increment, name varchar(32) DEFAULT NULL) auto_increment=256 "
+        String sql = "create table quniya4(name varchar(255) null,value varchar(255) null,id int not null,constraint quniya4_pk primary key (id));"
-                     + "ALTER TABLE yushitai_test.card_record ADD COLUMN remark2 varchar(255) DEFAULT NULL , ALGORITHM=inplace,LOCK=NONE;"
+                     + "alter table quniya4 modify id int not null first;";
-                     + "ALTER TABLE yushitai_test.card_record modify COLUMN name varchar(64) DEFAULT NULL , ALGORITHM=copy,LOCK=SHARED; "
-                     + "alter table yushitai_test.card_record add index index_name(name) ;"
-                     + "alter table yushitai_test.card_record add index index_name(name) ;"
-                     + "alter table yushitai_test.card_record add Constraint pk_id PRIMARY KEY (id);"
-                     + "alter table yushitai_test.card_record add Constraint pk_id PRIMARY KEY (id);"
-                     + "alter table yushitai_test.card_record add Constraint UNIQUE index uk_name(name);"
-                     + "alter table yushitai_test.card_record add Constraint UNIQUE index uk_name(name);";
         repository.console(sql);
 
-        repository.setDefaultSchema("yushitai_test");
+        repository.setDefaultSchema("test");
-        SchemaObject table = repository.findTable("card_record");
+        SchemaObject table = repository.findTable("quniya4");
         System.out.println(table.getStatement().toString());
     }
 }

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta_DDL_Test.java

@@ -18,7 +18,7 @@ import com.alibaba.otter.canal.parse.inbound.TableMeta;
  * @author agapple 2017年8月1日 下午7:15:54
  */
 @RunWith(SpringJUnit4ClassRunner.class)
-@ContextConfiguration(locations = { "/tsdb/mysql-tsdb.xml" })
+@ContextConfiguration(locations = { "/tsdb/h2-tsdb.xml" })
 public class MemoryTableMeta_DDL_Test {
 
     @Test
@@ -52,4 +52,18 @@ public class MemoryTableMeta_DDL_Test {
         Assert.assertEquals(meta.getFieldMetaByName("id").isKey(), true);
         Assert.assertEquals(meta.getFieldMetaByName("name").isUnique(), true);
     }
+
+    @Test
+    public void test3() throws Throwable {
+        MemoryTableMeta memoryTableMeta = new MemoryTableMeta();
+        URL url = Thread.currentThread().getContextClassLoader().getResource("dummy.txt");
+        File dummyFile = new File(url.getFile());
+        File create = new File(dummyFile.getParent() + "/ddl", "ddl_test3.sql");
+        String sql = StringUtils.join(IOUtils.readLines(new FileInputStream(create)), "\n");
+        memoryTableMeta.apply(null, "test", sql, null);
+
+        TableMeta meta = memoryTableMeta.find("test", "quniya4");
+        System.out.println(meta);
+        Assert.assertTrue(meta.getFields().get(0).getColumnName().equalsIgnoreCase("id"));
+    }
 }

parse/src/test/resources/ddl/ddl_test3.sql

@@ -0,0 +1,3 @@
+create table quniya4(name varchar(255) null,value varchar(255) null,id int not null,constraint quniya4_pk primary key (id));
+alter table quniya4 modify id int not null first;
+alter table quniya4 modify id int auto_increment;

pom.xml

@@ -4,7 +4,7 @@
     <artifactId>canal</artifactId>
     <packaging>pom</packaging>
     <name>canal module for otter ${project.version}</name>
-    <version>1.1.3-SNAPSHOT</version>
+    <version>1.1.3</version>
     <url>https://github.com/alibaba/canal</url>
     <parent>
         <groupId>org.sonatype.oss</groupId>
@@ -246,7 +246,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>com.alibaba</groupId>
@@ -381,7 +381,6 @@
                 </configuration>
             </plugin>
             <!-- javadoc -->
-            <!--
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-javadoc-plugin</artifactId>
@@ -400,7 +399,7 @@
                   <additionalparam>-Xdoclint:none</additionalparam>
                 </configuration>
             </plugin>
-            -->
+            
         </plugins>
         <sourceDirectory>src/main/java</sourceDirectory>
         <testSourceDirectory>src/test/java</testSourceDirectory>

prometheus/pom.xml

@@ -3,13 +3,13 @@
     <parent>
         <artifactId>canal</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.3-SNAPSHOT</version>
+        <version>1.1.3</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
     <groupId>com.alibaba.otter</groupId>
     <artifactId>canal.prometheus</artifactId>
-    <version>1.1.3-SNAPSHOT</version>
+    <version>1.1.3</version>
     <name>canal prometheus module for otter ${project.version}</name>
     <dependencies>
         <dependency>

protocol/pom.xml

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

server/pom.xml

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

server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java

@@ -36,8 +36,11 @@ public class MQMessageUtils {
 
                                                                                  public List<PartitionData> apply(String pkHashConfigs) {
                                                                                      List<PartitionData> datas = Lists.newArrayList();
-                                                                                     String[] pkHashConfigArray = StringUtils.split(pkHashConfigs,
+
-                                                                                         ",");
+                                                                                     String[] pkHashConfigArray = StringUtils.split(StringUtils.replace(pkHashConfigs,
+                                                                                         ",",
+                                                                                         ";"),
+                                                                                         ";");
                                                                                      // schema.table:id^name
                                                                                      for (String pkHashConfig : pkHashConfigArray) {
                                                                                          PartitionData data = new PartitionData();
@@ -75,8 +78,10 @@ public class MQMessageUtils {
 
                                                                                  public List<DynamicTopicData> apply(String pkHashConfigs) {
                                                                                      List<DynamicTopicData> datas = Lists.newArrayList();
-                                                                                     String[] dynamicTopicArray = StringUtils.split(pkHashConfigs,
+                                                                                     String[] dynamicTopicArray = StringUtils.split(StringUtils.replace(pkHashConfigs,
-                                                                                         ",");
+                                                                                         ",",
+                                                                                         ";"),
+                                                                                         ";");
                                                                                      // schema.table
                                                                                      for (String dynamicTopic : dynamicTopicArray) {
                                                                                          DynamicTopicData data = new DynamicTopicData();
@@ -216,7 +221,7 @@ public class MQMessageUtils {
                         partitionEntries[pkHash].add(entry);
                     } else {
                         for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
-                            int hashCode = table.hashCode();
+                            int hashCode = database.hashCode();
                             if (hashMode.autoPkHash) {
                                 // isEmpty use default pkNames
                                 for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
@@ -432,7 +437,7 @@ public class MQMessageUtils {
 
                     int idx = 0;
                     for (Map<String, String> row : flatMessage.getData()) {
-                        int hashCode = table.hashCode();
+                        int hashCode = database.hashCode();
                         for (String pkName : pkNames) {
                             String value = row.get(pkName);
                             if (value == null) {
@@ -509,7 +514,7 @@ public class MQMessageUtils {
     }
 
     private static Set<String> matchTopics(String name, String dynamicTopicConfigs) {
-        String[] router = StringUtils.split(dynamicTopicConfigs, ';');
+        String[] router = StringUtils.split(StringUtils.replace(dynamicTopicConfigs, ",", ";"), ";");
         Set<String> topics = new HashSet<>();
         for (String item : router) {
             int i = item.indexOf(":");

server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java

@@ -1,5 +1,6 @@
 package com.alibaba.otter.canal.kafka;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
@@ -138,7 +139,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
     private void send(MQProperties.CanalDestination canalDestination, String topicName, Message message)
                                                                                                         throws Exception {
         if (!kafkaProperties.getFlatMessage()) {
-            ProducerRecord<String, Message> record = null;
+            List<ProducerRecord<String, Message>> records = new ArrayList<ProducerRecord<String, Message>>();
             if (canalDestination.getPartitionHash() != null && !canalDestination.getPartitionHash().isEmpty()) {
                 Message[] messages = MQMessageUtils.messagePartition(message,
                     canalDestination.getPartitionsNum(),
@@ -147,18 +148,16 @@ public class CanalKafkaProducer implements CanalMQProducer {
                 for (int i = 0; i < length; i++) {
                     Message messagePartition = messages[i];
                     if (messagePartition != null) {
-                        record = new ProducerRecord<>(topicName, i, null, messagePartition);
+                        records.add(new ProducerRecord<String, Message>(topicName, i, null, messagePartition));
                     }
                 }
             } else {
                 final int partition = canalDestination.getPartition() != null ? canalDestination.getPartition() : 0;
-                record = new ProducerRecord<>(topicName, partition, null, message);
+                records.add(new ProducerRecord<String, Message>(topicName, partition, null, message));
             }
 
-            if (record != null) {
+            if (!records.isEmpty()) {
-                if (kafkaProperties.getTransaction()) {
+                for (ProducerRecord<String, Message> record : records) {
-                    producer.send(record).get();
-                } else {
                     producer.send(record).get();
                 }
 

server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java

@@ -107,11 +107,10 @@ public class CanalRocketMQProducer implements CanalMQProducer {
                                     @Override
                                     public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                                         if (index > mqs.size()) {
-                                            throw new CanalServerException("partition number is error,config num:"
+                                            return mqs.get(index % mqs.size());
-                                                                           + destination.getPartitionsNum()
+                                        } else {
-                                                                           + ", mq num: " + mqs.size());
+                                            return mqs.get(index);
                                         }
-                                        return mqs.get(index);
                                     }
                                 }, null);
                             } catch (Exception e) {
@@ -134,7 +133,11 @@ public class CanalRocketMQProducer implements CanalMQProducer {
 
                         @Override
                         public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
-                            return mqs.get(partition);
+                            if (partition > mqs.size()) {
+                                return mqs.get(partition % mqs.size());
+                            } else {
+                                return mqs.get(partition);
+                            }
                         }
                     }, null);
                 }
@@ -168,11 +171,10 @@ public class CanalRocketMQProducer implements CanalMQProducer {
                                         @Override
                                         public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                                             if (index > mqs.size()) {
-                                                throw new CanalServerException("partition number is error,config num:"
+                                                return mqs.get(index % mqs.size());
-                                                                               + destination.getPartitionsNum()
+                                            } else {
-                                                                               + ", mq num: " + mqs.size());
+                                                return mqs.get(index);
                                             }
-                                            return mqs.get(index);
                                         }
                                     }, null);
                                 } catch (Exception e) {
@@ -196,7 +198,11 @@ public class CanalRocketMQProducer implements CanalMQProducer {
 
                                 @Override
                                 public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
-                                    return mqs.get(partition);
+                                    if (partition > mqs.size()) {
+                                        return mqs.get(partition % mqs.size());
+                                    } else {
+                                        return mqs.get(partition);
+                                    }
                                 }
                             }, null);
                         } catch (Exception e) {

server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java

@@ -53,9 +53,6 @@ public class CanalMQStarter {
                 System.setProperty("canal.instance.filter.transaction.entry", "true");
             }
 
-            // 设置为raw避免ByteString->Entry的二次解析
-            System.setProperty("canal.instance.memory.rawEntry", "false");
-
             canalServer = CanalServerWithEmbedded.instance();
 
             // 对应每个instance启动一个worker线程

sink/pom.xml

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

store/pom.xml

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