feat 元数据与target数据源不一致时 进行兼容处理 (#5168)

* feat 元数据与target数据源不一致时 进行兼容处理

* feat 元数据与target数据源不一致时 进行兼容处理  优化代码

* Feature #FR-761 ddl执行失败会造成后续未执行的dml丢弃 针对ddl exec failed catch the ex

admin/admin-ui/pom.xml

@@ -3,7 +3,7 @@
   <parent>
     <artifactId>canal.admin</artifactId>
     <groupId>com.alibaba.otter</groupId>
-    <version>1.1.7-SNAPSHOT</version>
+    <version>1.1.7</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 

admin/admin-web/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.admin</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>canal.admin-web</artifactId>
@@ -30,6 +30,8 @@
         <dependency>
             <groupId>mysql</groupId>
             <artifactId>mysql-connector-java</artifactId>
+            <!-- 支持arm64需要mysql 8.0,升级为8.0的新版驱动 -->
+            <version>8.0.33</version>
         </dependency>
         <dependency>
             <groupId>com.github.ben-manes.caffeine</groupId>

admin/admin-web/src/main/bin/startup.sh

@@ -59,7 +59,7 @@ esac
 JavaVersion=`$JAVA -version 2>&1 |awk 'NR==1{ gsub(/"/,""); print $3 }' | awk  -F '.' '{print $1}'`
 str=`file -L $JAVA | grep 64-bit`
 
-JAVA_OPTS="$JAVA_OPTS -Xss256k -XX:+AggressiveOpts -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"
+JAVA_OPTS="$JAVA_OPTS -Xss1m -XX:+AggressiveOpts -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"
 if [ $JavaVersion -ge 11 ] ; then
   #JAVA_OPTS="$JAVA_OPTS -Xlog:gc*:$base_log/gc.log:time "
   JAVA_OPTS="$JAVA_OPTS"

admin/pom.xml

@@ -3,13 +3,13 @@
     <parent>
         <groupId>com.alibaba.otter</groupId>
         <artifactId>canal</artifactId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.alibaba.otter</groupId>
     <artifactId>canal.admin</artifactId>
-    <version>1.1.7-SNAPSHOT</version>
+    <version>1.1.7</version>
     <packaging>pom</packaging>
     <name>canal admin module for otter ${project.version}</name>
     <modules>

client-adapter/common/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
@@ -54,6 +54,11 @@
             <artifactId>junit</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.yaml</groupId>
+            <artifactId>snakeyaml</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>com.diffblue</groupId>
             <artifactId>deeptestutils</artifactId>

client-adapter/common/src/test/java/com/alibaba/otter/canal/client/adapter/support/YamlUtilsTest.java

@@ -1,14 +1,17 @@
 package com.alibaba.otter.canal.client.adapter.support;
 
+import org.junit.Assert;
+import org.junit.Test;
+import org.springframework.beans.factory.annotation.Value;
+
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Properties;
 
-import org.junit.Assert;
-import org.junit.Test;
-
 public class YamlUtilsTest {
 
+
+
     @Test
     public void testLoadConfigToYml() {
         String configStr="dataSourceKey: defaultDS\n"
@@ -17,6 +20,7 @@ public class YamlUtilsTest {
             + "outerAdapterKey: mysql1\n"
             + "concurrent: true\n"
             + "dbMapping:\n"
+            + "  _id: _id\n"
             + "  database: mytest\n"
             + "  table: user\n"
             + "  targetTable: mytest2.user\n"
@@ -35,6 +39,7 @@ public class YamlUtilsTest {
         MappingConfig config = YamlUtils.ymlToObj(null, configStr, MappingConfig.class, null, new Properties());
 
         Assert.assertNotNull(config);
+        Assert.assertEquals(config.getDbMapping().getId(), "_id");
         Assert.assertEquals(config.getDestination(), "example");
         Assert.assertEquals(config.getOuterAdapterKey(), "mysql1");
         Assert.assertEquals(config.getDbMapping().getDatabase(), "mytest");
@@ -104,6 +109,9 @@ public class YamlUtilsTest {
     }
 
     private static class DbMapping {
+
+        @Value("${_id}")
+        private String id ;
         private boolean             mirrorDb        = false;                 // 是否镜像库
         private String              database;                                // 数据库名或schema名
         private String              table;                                   // 表名
@@ -225,5 +233,13 @@ public class YamlUtilsTest {
         public void setAllMapColumns(Map<String, String> allMapColumns) {
             this.allMapColumns = allMapColumns;
         }
+
+        public String getId() {
+            return id;
+        }
+
+        public void setId(String id) {
+            this.id = id;
+        }
     }
 }

client-adapter/es6x/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/ES6xAdapter.java

@@ -59,12 +59,12 @@ public class ES6xAdapter extends ESAdapter {
     public Map<String, Object> count(String task) {
         ESSyncConfig config = esSyncConfig.get(task);
         ESSyncConfig.ESMapping mapping = config.getEsMapping();
-        SearchResponse response = this.esConnection.new ESSearchRequest(mapping.get_index(), mapping.get_type()).size(0)
+        SearchResponse response = this.esConnection.new ESSearchRequest(mapping.getIndex(), mapping.getType()).size(0)
             .getResponse();
 
         long rowCount = response.getHits().getTotalHits();
         Map<String, Object> res = new LinkedHashMap<>();
-        res.put("esIndex", mapping.get_index());
+        res.put("esIndex", mapping.getIndex());
         res.put("count", rowCount);
         return res;
     }

client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/etl/ESEtlService.java

@@ -51,7 +51,7 @@ public class ESEtlService extends AbstractEtlService {
 
     public EtlResult importData(List<String> params) {
         ESMapping mapping = config.getEsMapping();
-        logger.info("start etl to import data to index: {}", mapping.get_index());
+        logger.info("start etl to import data to index: {}", mapping.getIndex());
         String sql = mapping.getSql();
         return importData(sql, params);
     }
@@ -78,7 +78,7 @@ public class ESEtlService extends AbstractEtlService {
                             }
 
                             // 如果是主键字段则不插入
-                if (fieldItem.getFieldName().equals(mapping.get_id())) {
+                if (fieldItem.getFieldName().equals(mapping.getId())) {
                     idVal = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
                 } else {
                     Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
@@ -117,8 +117,8 @@ public class ESEtlService extends AbstractEtlService {
             if (idVal != null) {
                 String parentVal = (String) esFieldData.remove("$parent_routing");
                 if (mapping.isUpsert()) {
-                    ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.get_index(),
-                        mapping.get_type(),
+                    ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.getIndex(),
+                        mapping.getType(),
                         idVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
 
                     if (StringUtils.isNotEmpty(parentVal)) {
@@ -127,8 +127,8 @@ public class ESEtlService extends AbstractEtlService {
 
                     esBulkRequest.add(esUpdateRequest);
                 } else {
-                    ESIndexRequest esIndexRequest = this.esConnection.new ES6xIndexRequest(mapping.get_index(),
-                        mapping.get_type(),
+                    ESIndexRequest esIndexRequest = this.esConnection.new ES6xIndexRequest(mapping.getIndex(),
+                        mapping.getType(),
                         idVal.toString()).setSource(esFieldData);
                     if (StringUtils.isNotEmpty(parentVal)) {
                         esIndexRequest.setRouting(parentVal);
@@ -137,12 +137,12 @@ public class ESEtlService extends AbstractEtlService {
                 }
             } else {
                 idVal = esFieldData.get(mapping.getPk());
-                ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index(),
-                    mapping.get_type()).setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)).size(10000);
+                ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex(),
+                    mapping.getType()).setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)).size(10000);
                 SearchResponse response = esSearchRequest.getResponse();
                 for (SearchHit hit : response.getHits()) {
-                    ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.get_index(),
-                        mapping.get_type(),
+                    ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.getIndex(),
+                        mapping.getType(),
                         hit.getId()).setDoc(esFieldData);
                     esBulkRequest.add(esUpdateRequest);
                 }
@@ -160,7 +160,7 @@ public class ESEtlService extends AbstractEtlService {
                         (System.currentTimeMillis() - batchBegin),
                         (System.currentTimeMillis() - esBatchBegin),
                         esBulkRequest.numberOfActions(),
-                        mapping.get_index());
+                        mapping.getIndex());
                 }
                 batchBegin = System.currentTimeMillis();
                 esBulkRequest.resetBulk();
@@ -180,12 +180,12 @@ public class ESEtlService extends AbstractEtlService {
                     (System.currentTimeMillis() - batchBegin),
                     (System.currentTimeMillis() - esBatchBegin),
                     esBulkRequest.numberOfActions(),
-                    mapping.get_index());
+                    mapping.getIndex());
             }
         }
     } catch (Exception e) {
         logger.error(e.getMessage(), e);
-        errMsg.add(mapping.get_index() + " etl failed! ==>" + e.getMessage());
+        errMsg.add(mapping.getIndex() + " etl failed! ==>" + e.getMessage());
         throw new RuntimeException(e);
     }
     return count;

client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/support/ES6xTemplate.java

@@ -66,19 +66,19 @@ public class ES6xTemplate implements ESTemplate {
 
     @Override
     public void insert(ESSyncConfig.ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
-        if (mapping.get_id() != null) {
+        if (mapping.getId() != null) {
             String parentVal = (String) esFieldData.remove("$parent_routing");
             if (mapping.isUpsert()) {
-                ESUpdateRequest updateRequest = esConnection.new ES6xUpdateRequest(mapping.get_index(),
-                    mapping.get_type(),
+                ESUpdateRequest updateRequest = esConnection.new ES6xUpdateRequest(mapping.getIndex(),
+                    mapping.getType(),
                     pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
                 if (StringUtils.isNotEmpty(parentVal)) {
                     updateRequest.setRouting(parentVal);
                 }
                 getBulk().add(updateRequest);
             } else {
-                ESIndexRequest indexRequest = esConnection.new ES6xIndexRequest(mapping.get_index(),
-                    mapping.get_type(),
+                ESIndexRequest indexRequest = esConnection.new ES6xIndexRequest(mapping.getIndex(),
+                    mapping.getType(),
                     pkVal.toString()).setSource(esFieldData);
                 if (StringUtils.isNotEmpty(parentVal)) {
                     indexRequest.setRouting(parentVal);
@@ -87,13 +87,13 @@ public class ES6xTemplate implements ESTemplate {
             }
             commitBulk();
         } else {
-            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index(),
-                mapping.get_type()).setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)).size(10000);
+            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex(),
+                mapping.getType()).setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)).size(10000);
             SearchResponse response = esSearchRequest.getResponse();
 
             for (SearchHit hit : response.getHits()) {
-                ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.get_index(),
-                    mapping.get_type(),
+                ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.getIndex(),
+                    mapping.getType(),
                     hit.getId()).setDoc(esFieldData);
                 getBulk().add(esUpdateRequest);
                 commitBulk();
@@ -150,19 +150,19 @@ public class ES6xTemplate implements ESTemplate {
 
     @Override
     public void delete(ESSyncConfig.ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
-        if (mapping.get_id() != null) {
-            ESDeleteRequest esDeleteRequest = this.esConnection.new ES6xDeleteRequest(mapping.get_index(),
-                mapping.get_type(),
+        if (mapping.getId() != null) {
+            ESDeleteRequest esDeleteRequest = this.esConnection.new ES6xDeleteRequest(mapping.getIndex(),
+                mapping.getType(),
                 pkVal.toString());
             getBulk().add(esDeleteRequest);
             commitBulk();
         } else {
-            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index(),
-                mapping.get_type()).setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)).size(10000);
+            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex(),
+                mapping.getType()).setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)).size(10000);
             SearchResponse response = esSearchRequest.getResponse();
             for (SearchHit hit : response.getHits()) {
-                ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.get_index(),
-                    mapping.get_type(),
+                ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.getIndex(),
+                    mapping.getType(),
                     hit.getId()).setDoc(esFieldData);
                 getBulk().add(esUpdateRequest);
                 commitBulk();
@@ -207,7 +207,7 @@ public class ES6xTemplate implements ESTemplate {
     public Object getESDataFromRS(ESSyncConfig.ESMapping mapping, ResultSet resultSet,
                                   Map<String, Object> esFieldData) throws SQLException {
         SchemaItem schemaItem = mapping.getSchemaItem();
-        String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+        String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
             Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
@@ -216,7 +216,7 @@ public class ES6xTemplate implements ESTemplate {
                 resultIdVal = value;
             }
 
-            if (!fieldItem.getFieldName().equals(mapping.get_id())
+            if (!fieldItem.getFieldName().equals(mapping.getId())
                     && !mapping.getSkips().contains(fieldItem.getFieldName())) {
                 esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
             }
@@ -231,7 +231,7 @@ public class ES6xTemplate implements ESTemplate {
     @Override
     public Object getIdValFromRS(ESSyncConfig.ESMapping mapping, ResultSet resultSet) throws SQLException {
         SchemaItem schemaItem = mapping.getSchemaItem();
-        String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+        String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
             Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
@@ -248,7 +248,7 @@ public class ES6xTemplate implements ESTemplate {
     public Object getESDataFromRS(ESSyncConfig.ESMapping mapping, ResultSet resultSet, Map<String, Object> dmlOld,
                                   Map<String, Object> esFieldData) throws SQLException {
         SchemaItem schemaItem = mapping.getSchemaItem();
-        String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+        String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
             if (fieldItem.getFieldName().equals(idFieldName)) {
@@ -294,7 +294,7 @@ public class ES6xTemplate implements ESTemplate {
     public Object getESDataFromDmlData(ESSyncConfig.ESMapping mapping, Map<String, Object> dmlData,
                                        Map<String, Object> esFieldData) {
         SchemaItem schemaItem = mapping.getSchemaItem();
-        String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+        String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
             String columnName = fieldItem.getColumnItems().iterator().next().getColumnName();
@@ -304,7 +304,7 @@ public class ES6xTemplate implements ESTemplate {
                 resultIdVal = value;
             }
 
-            if (!fieldItem.getFieldName().equals(mapping.get_id())
+            if (!fieldItem.getFieldName().equals(mapping.getId())
                     && !mapping.getSkips().contains(fieldItem.getFieldName())) {
                 esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
             }
@@ -319,7 +319,7 @@ public class ES6xTemplate implements ESTemplate {
     public Object getESDataFromDmlData(ESSyncConfig.ESMapping mapping,String owner, Map<String, Object> dmlData,
                                        Map<String, Object> dmlOld, Map<String, Object> esFieldData) {
         SchemaItem schemaItem = mapping.getSchemaItem();
-        String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+        String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
             ColumnItem columnItem = fieldItem.getColumnItems().iterator().next();
@@ -353,19 +353,19 @@ public class ES6xTemplate implements ESTemplate {
     }
 
     private void append4Update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
-        if (mapping.get_id() != null) {
+        if (mapping.getId() != null) {
             String parentVal = (String) esFieldData.remove("$parent_routing");
             if (mapping.isUpsert()) {
-                ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.get_index(),
-                    mapping.get_type(),
+                ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.getIndex(),
+                    mapping.getType(),
                     pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
                 if (StringUtils.isNotEmpty(parentVal)) {
                     esUpdateRequest.setRouting(parentVal);
                 }
                 getBulk().add(esUpdateRequest);
             } else {
-                ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.get_index(),
-                    mapping.get_type(),
+                ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.getIndex(),
+                    mapping.getType(),
                     pkVal.toString()).setDoc(esFieldData);
                 if (StringUtils.isNotEmpty(parentVal)) {
                     esUpdateRequest.setRouting(parentVal);
@@ -373,12 +373,12 @@ public class ES6xTemplate implements ESTemplate {
                 getBulk().add(esUpdateRequest);
             }
         } else {
-            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index(),
-                mapping.get_type()).setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)).size(10000);
+            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex(),
+                mapping.getType()).setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)).size(10000);
             SearchResponse response = esSearchRequest.getResponse();
             for (SearchHit hit : response.getHits()) {
-                ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.get_index(),
-                    mapping.get_type(),
+                ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.getIndex(),
+                    mapping.getType(),
                     hit.getId()).setDoc(esFieldData);
                 getBulk().add(esUpdateRequest);
             }
@@ -394,15 +394,15 @@ public class ES6xTemplate implements ESTemplate {
      */
     @SuppressWarnings("unchecked")
     private String getEsType(ESMapping mapping, String fieldName) {
-        String key = mapping.get_index() + "-" + mapping.get_type();
+        String key = mapping.getIndex() + "-" + mapping.getType();
         Map<String, String> fieldType = esFieldTypes.get(key);
         if (fieldType != null) {
             return fieldType.get(fieldName);
         } else {
-            MappingMetaData mappingMetaData = esConnection.getMapping(mapping.get_index(), mapping.get_type());
+            MappingMetaData mappingMetaData = esConnection.getMapping(mapping.getIndex(), mapping.getType());
 
             if (mappingMetaData == null) {
-                throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.get_index());
+                throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.getIndex());
             }
 
             fieldType = new LinkedHashMap<>();

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

@@ -29,9 +29,9 @@ public class ConfigLoadTest {
         Assert.assertNotNull(config);
         Assert.assertEquals("defaultDS", config.getDataSourceKey());
         ESSyncConfig.ESMapping esMapping = config.getEsMapping();
-        Assert.assertEquals("mytest_user", esMapping.get_index());
-        Assert.assertEquals("_doc", esMapping.get_type());
-        Assert.assertEquals("id", esMapping.get_id());
+        Assert.assertEquals("mytest_user", esMapping.getIndex());
+        Assert.assertEquals("_doc", esMapping.getType());
+        Assert.assertEquals("id", esMapping.getId());
         Assert.assertNotNull(esMapping.getSql());
 
         // Map<String, List<ESSyncConfig>> dbTableEsSyncConfig =

client-adapter/es7x/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/ES7xAdapter.java

@@ -59,11 +59,11 @@ public class ES7xAdapter extends ESAdapter {
     public Map<String, Object> count(String task) {
         ESSyncConfig config = esSyncConfig.get(task);
         ESSyncConfig.ESMapping mapping = config.getEsMapping();
-        SearchResponse response = this.esConnection.new ESSearchRequest(mapping.get_index()).size(0).getResponse();
+        SearchResponse response = this.esConnection.new ESSearchRequest(mapping.getIndex()).size(0).getResponse();
 
         long rowCount = response.getHits().getTotalHits().value;
         Map<String, Object> res = new LinkedHashMap<>();
-        res.put("esIndex", mapping.get_index());
+        res.put("esIndex", mapping.getIndex());
         res.put("count", rowCount);
         return res;
     }

client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/etl/ESEtlService.java

@@ -51,7 +51,7 @@ public class ESEtlService extends AbstractEtlService {
 
     public EtlResult importData(List<String> params) {
         ESMapping mapping = config.getEsMapping();
-        logger.info("start etl to import data to index: {}", mapping.get_index());
+        logger.info("start etl to import data to index: {}", mapping.getIndex());
         String sql = mapping.getSql();
         return importData(sql, params);
     }
@@ -78,7 +78,7 @@ public class ESEtlService extends AbstractEtlService {
                             }
 
                             // 如果是主键字段则不插入
-                            if (fieldItem.getFieldName().equals(mapping.get_id())) {
+                            if (fieldItem.getFieldName().equals(mapping.getId())) {
                                 idVal = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
                             } else {
                                 Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
@@ -118,7 +118,7 @@ public class ESEtlService extends AbstractEtlService {
                             String parentVal = (String) esFieldData.remove("$parent_routing");
                             if (mapping.isUpsert()) {
                                 ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(
-                                    mapping.get_index(),
+                                    mapping.getIndex(),
                                     idVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
 
                                 if (StringUtils.isNotEmpty(parentVal)) {
@@ -128,7 +128,7 @@ public class ESEtlService extends AbstractEtlService {
                                 esBulkRequest.add(esUpdateRequest);
                             } else {
                                 ESIndexRequest esIndexRequest = this.esConnection.new ES7xIndexRequest(
-                                    mapping.get_index(),
+                                    mapping.getIndex(),
                                     idVal.toString()).setSource(esFieldData);
                                 if (StringUtils.isNotEmpty(parentVal)) {
                                     esIndexRequest.setRouting(parentVal);
@@ -137,13 +137,13 @@ public class ESEtlService extends AbstractEtlService {
                             }
                         } else {
                             idVal = esFieldData.get(mapping.getPk());
-                            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index())
+                            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex())
                                 .setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal))
                                 .size(10000);
                             SearchResponse response = esSearchRequest.getResponse();
                             for (SearchHit hit : response.getHits()) {
                                 ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(
-                                    mapping.get_index(),
+                                    mapping.getIndex(),
                                     hit.getId()).setDoc(esFieldData);
                                 esBulkRequest.add(esUpdateRequest);
                             }
@@ -162,7 +162,7 @@ public class ESEtlService extends AbstractEtlService {
                                     (System.currentTimeMillis() - batchBegin),
                                     (System.currentTimeMillis() - esBatchBegin),
                                     esBulkRequest.numberOfActions(),
-                                    mapping.get_index());
+                                    mapping.getIndex());
                             }
                             batchBegin = System.currentTimeMillis();
                             esBulkRequest.resetBulk();
@@ -182,12 +182,12 @@ public class ESEtlService extends AbstractEtlService {
                                 (System.currentTimeMillis() - batchBegin),
                                 (System.currentTimeMillis() - esBatchBegin),
                                 esBulkRequest.numberOfActions(),
-                                mapping.get_index());
+                                mapping.getIndex());
                         }
                     }
                 } catch (Exception e) {
                     logger.error(e.getMessage(), e);
-                    errMsg.add(mapping.get_index() + " etl failed! ==>" + e.getMessage());
+                    errMsg.add(mapping.getIndex() + " etl failed! ==>" + e.getMessage());
                     throw new RuntimeException(e);
                 }
                 return count;

client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/support/ES7xTemplate.java

@@ -66,17 +66,17 @@ public class ES7xTemplate implements ESTemplate {
 
     @Override
     public void insert(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
-        if (mapping.get_id() != null) {
+        if (mapping.getId() != null) {
             String parentVal = (String) esFieldData.remove("$parent_routing");
             if (mapping.isUpsert()) {
-                ESUpdateRequest updateRequest = esConnection.new ES7xUpdateRequest(mapping.get_index(),
+                ESUpdateRequest updateRequest = esConnection.new ES7xUpdateRequest(mapping.getIndex(),
                     pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
                 if (StringUtils.isNotEmpty(parentVal)) {
                     updateRequest.setRouting(parentVal);
                 }
                 getBulk().add(updateRequest);
             } else {
-                ESIndexRequest indexRequest = esConnection.new ES7xIndexRequest(mapping.get_index(), pkVal.toString())
+                ESIndexRequest indexRequest = esConnection.new ES7xIndexRequest(mapping.getIndex(), pkVal.toString())
                     .setSource(esFieldData);
                 if (StringUtils.isNotEmpty(parentVal)) {
                     indexRequest.setRouting(parentVal);
@@ -85,13 +85,13 @@ public class ES7xTemplate implements ESTemplate {
             }
             commitBulk();
         } else {
-            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index())
+            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex())
                 .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
                 .size(10000);
             SearchResponse response = esSearchRequest.getResponse();
 
             for (SearchHit hit : response.getHits()) {
-                ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(),
+                ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.getIndex(),
                     hit.getId()).setDoc(esFieldData);
                 getBulk().add(esUpdateRequest);
                 commitBulk();
@@ -148,18 +148,18 @@ public class ES7xTemplate implements ESTemplate {
 
     @Override
     public void delete(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
-        if (mapping.get_id() != null) {
-            ESDeleteRequest esDeleteRequest = this.esConnection.new ES7xDeleteRequest(mapping.get_index(),
+        if (mapping.getId() != null) {
+            ESDeleteRequest esDeleteRequest = this.esConnection.new ES7xDeleteRequest(mapping.getIndex(),
                 pkVal.toString());
             getBulk().add(esDeleteRequest);
             commitBulk();
         } else {
-            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index())
+            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex())
                 .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
                 .size(10000);
             SearchResponse response = esSearchRequest.getResponse();
             for (SearchHit hit : response.getHits()) {
-                ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(),
+                ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.getIndex(),
                     hit.getId()).setDoc(esFieldData);
                 getBulk().add(esUpdateRequest);
                 commitBulk();
@@ -204,7 +204,7 @@ public class ES7xTemplate implements ESTemplate {
     public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet,
                                   Map<String, Object> esFieldData) throws SQLException {
         SchemaItem schemaItem = mapping.getSchemaItem();
-        String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+        String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
             Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
@@ -213,7 +213,7 @@ public class ES7xTemplate implements ESTemplate {
                 resultIdVal = value;
             }
 
-            if (!fieldItem.getFieldName().equals(mapping.get_id())
+            if (!fieldItem.getFieldName().equals(mapping.getId())
                 && !mapping.getSkips().contains(fieldItem.getFieldName())) {
                 esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
             }
@@ -228,7 +228,7 @@ public class ES7xTemplate implements ESTemplate {
     @Override
     public Object getIdValFromRS(ESMapping mapping, ResultSet resultSet) throws SQLException {
         SchemaItem schemaItem = mapping.getSchemaItem();
-        String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+        String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
             Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
@@ -245,7 +245,7 @@ public class ES7xTemplate implements ESTemplate {
     public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map<String, Object> dmlOld,
                                   Map<String, Object> esFieldData) throws SQLException {
         SchemaItem schemaItem = mapping.getSchemaItem();
-        String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+        String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
             if (fieldItem.getFieldName().equals(idFieldName)) {
@@ -289,7 +289,7 @@ public class ES7xTemplate implements ESTemplate {
     @Override
     public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData, Map<String, Object> esFieldData) {
         SchemaItem schemaItem = mapping.getSchemaItem();
-        String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+        String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
             String columnName = fieldItem.getColumnItems().iterator().next().getColumnName();
@@ -299,7 +299,7 @@ public class ES7xTemplate implements ESTemplate {
                 resultIdVal = value;
             }
 
-            if (!fieldItem.getFieldName().equals(mapping.get_id())
+            if (!fieldItem.getFieldName().equals(mapping.getId())
                 && !mapping.getSkips().contains(fieldItem.getFieldName())) {
                 esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
             }
@@ -314,7 +314,7 @@ public class ES7xTemplate implements ESTemplate {
     public Object getESDataFromDmlData(ESMapping mapping,String owner, Map<String, Object> dmlData, Map<String, Object> dmlOld,
                                        Map<String, Object> esFieldData) {
         SchemaItem schemaItem = mapping.getSchemaItem();
-        String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+        String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
             ColumnItem columnItem = fieldItem.getColumnItems().iterator().next();
@@ -348,17 +348,17 @@ public class ES7xTemplate implements ESTemplate {
     }
 
     private void append4Update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
-        if (mapping.get_id() != null) {
+        if (mapping.getId() != null) {
             String parentVal = (String) esFieldData.remove("$parent_routing");
             if (mapping.isUpsert()) {
-                ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(),
+                ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.getIndex(),
                     pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
                 if (StringUtils.isNotEmpty(parentVal)) {
                     esUpdateRequest.setRouting(parentVal);
                 }
                 getBulk().add(esUpdateRequest);
             } else {
-                ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(),
+                ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.getIndex(),
                     pkVal.toString()).setDoc(esFieldData);
                 if (StringUtils.isNotEmpty(parentVal)) {
                     esUpdateRequest.setRouting(parentVal);
@@ -366,12 +366,12 @@ public class ES7xTemplate implements ESTemplate {
                 getBulk().add(esUpdateRequest);
             }
         } else {
-            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index())
+            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex())
                 .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
                 .size(10000);
             SearchResponse response = esSearchRequest.getResponse();
             for (SearchHit hit : response.getHits()) {
-                ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(),
+                ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.getIndex(),
                     hit.getId()).setDoc(esFieldData);
                 getBulk().add(esUpdateRequest);
             }
@@ -387,15 +387,15 @@ public class ES7xTemplate implements ESTemplate {
      */
     @SuppressWarnings("unchecked")
     private String getEsType(ESMapping mapping, String fieldName) {
-        String key = mapping.get_index() + "-" + mapping.get_type();
+        String key = mapping.getIndex() + "-" + mapping.getType();
         Map<String, String> fieldType = esFieldTypes.get(key);
         if (fieldType != null) {
             return fieldType.get(fieldName);
         } else {
-            MappingMetaData mappingMetaData = esConnection.getMapping(mapping.get_index());
+            MappingMetaData mappingMetaData = esConnection.getMapping(mapping.getIndex());
 
             if (mappingMetaData == null) {
-                throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.get_index());
+                throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.getIndex());
             }
 
             fieldType = new LinkedHashMap<>();

client-adapter/es8x/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"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+<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">
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
@@ -81,7 +80,7 @@
                             <tasks>
                                 <copy todir="${project.basedir}/../launcher/target/classes/es8" overwrite="true">
                                     <fileset dir="${project.basedir}/target/classes/es8" erroronmissingdir="true">
-                                        <include name="*.yml"/>
+                                        <include name="*.yml" />
                                     </fileset>
                                 </copy>
                             </tasks>

client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/ES8xAdapter.java

@@ -53,11 +53,11 @@ public class ES8xAdapter extends ESAdapter {
     public Map<String, Object> count(String task) {
         ESSyncConfig config = esSyncConfig.get(task);
         ESSyncConfig.ESMapping mapping = config.getEsMapping();
-        SearchResponse response = this.esConnection.new ESSearchRequest(mapping.get_index()).size(0).getResponse();
+        SearchResponse response = this.esConnection.new ESSearchRequest(mapping.getIndex()).size(0).getResponse();
 
         long rowCount = response.getHits().getTotalHits().value;
         Map<String, Object> res = new LinkedHashMap<>();
-        res.put("esIndex", mapping.get_index());
+        res.put("esIndex", mapping.getIndex());
         res.put("count", rowCount);
         return res;
     }

client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/etl/ESEtlService.java

@@ -51,7 +51,7 @@ public class ESEtlService extends AbstractEtlService {
 
     public EtlResult importData(List<String> params) {
         ESMapping mapping = config.getEsMapping();
-        logger.info("start etl to import data to index: {}", mapping.get_index());
+        logger.info("start etl to import data to index: {}", mapping.getIndex());
         String sql = mapping.getSql();
         return importData(sql, params);
     }
@@ -78,7 +78,7 @@ public class ESEtlService extends AbstractEtlService {
                             }
 
                             // 如果是主键字段则不插入
-                            if (fieldItem.getFieldName().equals(mapping.get_id())) {
+                            if (fieldItem.getFieldName().equals(mapping.getId())) {
                                 idVal = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
                             } else {
                                 Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
@@ -118,7 +118,7 @@ public class ESEtlService extends AbstractEtlService {
                             String parentVal = (String) esFieldData.remove("$parent_routing");
                             if (mapping.isUpsert()) {
                                 ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(
-                                    mapping.get_index(),
+                                    mapping.getIndex(),
                                     idVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
 
                                 if (StringUtils.isNotEmpty(parentVal)) {
@@ -128,7 +128,7 @@ public class ESEtlService extends AbstractEtlService {
                                 esBulkRequest.add(esUpdateRequest);
                             } else {
                                 ESIndexRequest esIndexRequest = this.esConnection.new ES8xIndexRequest(
-                                    mapping.get_index(),
+                                    mapping.getIndex(),
                                     idVal.toString()).setSource(esFieldData);
                                 if (StringUtils.isNotEmpty(parentVal)) {
                                     esIndexRequest.setRouting(parentVal);
@@ -137,13 +137,13 @@ public class ESEtlService extends AbstractEtlService {
                             }
                         } else {
                             idVal = esFieldData.get(mapping.getPk());
-                            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index())
+                            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex())
                                 .setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal))
                                 .size(10000);
                             SearchResponse response = esSearchRequest.getResponse();
                             for (SearchHit hit : response.getHits()) {
                                 ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(
-                                    mapping.get_index(),
+                                    mapping.getIndex(),
                                     hit.getId()).setDoc(esFieldData);
                                 esBulkRequest.add(esUpdateRequest);
                             }
@@ -162,7 +162,7 @@ public class ESEtlService extends AbstractEtlService {
                                     (System.currentTimeMillis() - batchBegin),
                                     (System.currentTimeMillis() - esBatchBegin),
                                     esBulkRequest.numberOfActions(),
-                                    mapping.get_index());
+                                    mapping.getIndex());
                             }
                             batchBegin = System.currentTimeMillis();
                             esBulkRequest.resetBulk();
@@ -182,12 +182,12 @@ public class ESEtlService extends AbstractEtlService {
                                 (System.currentTimeMillis() - batchBegin),
                                 (System.currentTimeMillis() - esBatchBegin),
                                 esBulkRequest.numberOfActions(),
-                                mapping.get_index());
+                                mapping.getIndex());
                         }
                     }
                 } catch (Exception e) {
                     logger.error(e.getMessage(), e);
-                    errMsg.add(mapping.get_index() + " etl failed! ==>" + e.getMessage());
+                    errMsg.add(mapping.getIndex() + " etl failed! ==>" + e.getMessage());
                     throw new RuntimeException(e);
                 }
                 return count;

client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/support/ES8xTemplate.java

@@ -62,17 +62,17 @@ public class ES8xTemplate implements ESTemplate {
 
     @Override
     public void insert(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
-        if (mapping.get_id() != null) {
+        if (mapping.getId() != null) {
             String parentVal = (String) esFieldData.remove("$parent_routing");
             if (mapping.isUpsert()) {
-                ESUpdateRequest updateRequest = esConnection.new ES8xUpdateRequest(mapping.get_index(),
+                ESUpdateRequest updateRequest = esConnection.new ES8xUpdateRequest(mapping.getIndex(),
                     pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
                 if (StringUtils.isNotEmpty(parentVal)) {
                     updateRequest.setRouting(parentVal);
                 }
                 getBulk().add(updateRequest);
             } else {
-                ESIndexRequest indexRequest = esConnection.new ES8xIndexRequest(mapping.get_index(), pkVal.toString())
+                ESIndexRequest indexRequest = esConnection.new ES8xIndexRequest(mapping.getIndex(), pkVal.toString())
                     .setSource(esFieldData);
                 if (StringUtils.isNotEmpty(parentVal)) {
                     indexRequest.setRouting(parentVal);
@@ -81,13 +81,13 @@ public class ES8xTemplate implements ESTemplate {
             }
             commitBulk();
         } else {
-            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index())
+            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex())
                 .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
                 .size(10000);
             SearchResponse response = esSearchRequest.getResponse();
 
             for (SearchHit hit : response.getHits()) {
-                ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.get_index(),
+                ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.getIndex(),
                     hit.getId()).setDoc(esFieldData);
                 getBulk().add(esUpdateRequest);
                 commitBulk();
@@ -144,18 +144,18 @@ public class ES8xTemplate implements ESTemplate {
 
     @Override
     public void delete(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
-        if (mapping.get_id() != null) {
-            ESDeleteRequest esDeleteRequest = this.esConnection.new ES8xDeleteRequest(mapping.get_index(),
+        if (mapping.getId() != null) {
+            ESDeleteRequest esDeleteRequest = this.esConnection.new ES8xDeleteRequest(mapping.getIndex(),
                 pkVal.toString());
             getBulk().add(esDeleteRequest);
             commitBulk();
         } else {
-            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index())
+            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex())
                 .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
                 .size(10000);
             SearchResponse response = esSearchRequest.getResponse();
             for (SearchHit hit : response.getHits()) {
-                ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.get_index(),
+                ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.getIndex(),
                     hit.getId()).setDoc(esFieldData);
                 getBulk().add(esUpdateRequest);
                 commitBulk();
@@ -200,7 +200,7 @@ public class ES8xTemplate implements ESTemplate {
     public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet,
                                   Map<String, Object> esFieldData) throws SQLException {
         SchemaItem schemaItem = mapping.getSchemaItem();
-        String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+        String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
             Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
@@ -209,7 +209,7 @@ public class ES8xTemplate implements ESTemplate {
                 resultIdVal = value;
             }
 
-            if (!fieldItem.getFieldName().equals(mapping.get_id())
+            if (!fieldItem.getFieldName().equals(mapping.getId())
                 && !mapping.getSkips().contains(fieldItem.getFieldName())) {
                 esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
             }
@@ -224,7 +224,7 @@ public class ES8xTemplate implements ESTemplate {
     @Override
     public Object getIdValFromRS(ESMapping mapping, ResultSet resultSet) throws SQLException {
         SchemaItem schemaItem = mapping.getSchemaItem();
-        String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+        String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
             Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
@@ -241,7 +241,7 @@ public class ES8xTemplate implements ESTemplate {
     public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map<String, Object> dmlOld,
                                   Map<String, Object> esFieldData) throws SQLException {
         SchemaItem schemaItem = mapping.getSchemaItem();
-        String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+        String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
             if (fieldItem.getFieldName().equals(idFieldName)) {
@@ -286,7 +286,7 @@ public class ES8xTemplate implements ESTemplate {
     public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData,
                                        Map<String, Object> esFieldData) {
         SchemaItem schemaItem = mapping.getSchemaItem();
-        String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+        String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
             String columnName = fieldItem.getColumnItems().iterator().next().getColumnName();
@@ -296,7 +296,7 @@ public class ES8xTemplate implements ESTemplate {
                 resultIdVal = value;
             }
 
-            if (!fieldItem.getFieldName().equals(mapping.get_id())
+            if (!fieldItem.getFieldName().equals(mapping.getId())
                 && !mapping.getSkips().contains(fieldItem.getFieldName())) {
                 esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
             }
@@ -311,7 +311,7 @@ public class ES8xTemplate implements ESTemplate {
     public Object getESDataFromDmlData(ESMapping mapping, String owner, Map<String, Object> dmlData,
                                        Map<String, Object> dmlOld, Map<String, Object> esFieldData) {
         SchemaItem schemaItem = mapping.getSchemaItem();
-        String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+        String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
             ColumnItem columnItem = fieldItem.getColumnItems().iterator().next();
@@ -345,17 +345,17 @@ public class ES8xTemplate implements ESTemplate {
     }
 
     private void append4Update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
-        if (mapping.get_id() != null) {
+        if (mapping.getId() != null) {
             String parentVal = (String) esFieldData.remove("$parent_routing");
             if (mapping.isUpsert()) {
-                ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.get_index(),
+                ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.getIndex(),
                     pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
                 if (StringUtils.isNotEmpty(parentVal)) {
                     esUpdateRequest.setRouting(parentVal);
                 }
                 getBulk().add(esUpdateRequest);
             } else {
-                ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.get_index(),
+                ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.getIndex(),
                     pkVal.toString()).setDoc(esFieldData);
                 if (StringUtils.isNotEmpty(parentVal)) {
                     esUpdateRequest.setRouting(parentVal);
@@ -363,12 +363,12 @@ public class ES8xTemplate implements ESTemplate {
                 getBulk().add(esUpdateRequest);
             }
         } else {
-            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index())
+            ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex())
                 .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
                 .size(10000);
             SearchResponse response = esSearchRequest.getResponse();
             for (SearchHit hit : response.getHits()) {
-                ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.get_index(),
+                ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.getIndex(),
                     hit.getId()).setDoc(esFieldData);
                 getBulk().add(esUpdateRequest);
             }
@@ -384,15 +384,15 @@ public class ES8xTemplate implements ESTemplate {
      */
     @SuppressWarnings("unchecked")
     private String getEsType(ESMapping mapping, String fieldName) {
-        String key = mapping.get_index() + "-" + mapping.get_type();
+        String key = mapping.getIndex() + "-" + mapping.getType();
         Map<String, String> fieldType = esFieldTypes.get(key);
         if (fieldType != null) {
             return fieldType.get(fieldName);
         } else {
-            MappingMetadata mappingMetaData = esConnection.getMapping(mapping.get_index());
+            MappingMetadata mappingMetaData = esConnection.getMapping(mapping.getIndex());
 
             if (mappingMetaData == null) {
-                throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.get_index());
+                throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.getIndex());
             }
 
             fieldType = new LinkedHashMap<>();

client-adapter/escore/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

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

@@ -6,6 +6,7 @@ import java.util.List;
 import java.util.Map;
 
 import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
+import org.springframework.beans.factory.annotation.Value;
 
 /**
  * ES 映射配置
@@ -28,13 +29,13 @@ public class ESSyncConfig implements AdapterConfig {
     private String    esVersion = "es6";
 
     public void validate() {
-        if (esMapping._index == null) {
+        if (esMapping.index == null) {
             throw new NullPointerException("esMapping._index");
         }
-        if ("es6".equals(esVersion) && esMapping._type == null) {
+        if ("es6".equals(esVersion) && esMapping.type == null) {
             throw new NullPointerException("esMapping._type");
         }
-        if (esMapping._id == null && esMapping.getPk() == null) {
+        if (esMapping.id == null && esMapping.getPk() == null) {
             throw new NullPointerException("esMapping._id or esMapping.pk");
         }
         if (esMapping.sql == null) {
@@ -96,9 +97,14 @@ public class ESSyncConfig implements AdapterConfig {
 
     public static class ESMapping implements AdapterMapping {
 
-        private String                       _index;
-        private String                       _type;
-        private String                       _id;
+        @Value("${_index}")
+        private String index;
+
+        @Value("${_type}")
+        private String type;
+
+        @Value("${_id}")
+        private String id;
         private boolean                      upsert          = false;
         private String                       pk;
         private Map<String, RelationMapping> relations       = new LinkedHashMap<>();
@@ -114,28 +120,28 @@ public class ESSyncConfig implements AdapterConfig {
 
         private SchemaItem                   schemaItem;                             // sql解析结果模型
 
-        public String get_index() {
-            return _index;
+        public String getIndex() {
+            return index;
         }
 
-        public void set_index(String _index) {
-            this._index = _index;
+        public void setIndex(String index) {
+            this.index = index;
         }
 
-        public String get_type() {
-            return _type;
+        public String getType() {
+            return type;
         }
 
-        public void set_type(String _type) {
-            this._type = _type;
+        public void setType(String type) {
+            this.type = type;
         }
 
-        public String get_id() {
-            return _id;
+        public String getId() {
+            return id;
         }
 
-        public void set_id(String _id) {
-            this._id = _id;
+        public void setId(String id) {
+            this.id = id;
         }
 
         public boolean isUpsert() {

client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/SchemaItem.java

@@ -136,8 +136,8 @@ public class SchemaItem {
     }
 
     public FieldItem getIdFieldItem(ESSyncConfig.ESMapping mapping) {
-        if (mapping.get_id() != null) {
-            return getSelectFields().get(mapping.get_id());
+        if (mapping.getId() != null) {
+            return getSelectFields().get(mapping.getId());
         } else {
             return getSelectFields().get(mapping.getPk());
         }

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

@@ -54,13 +54,13 @@ public class ESSyncService {
             for (ESSyncConfig config : esSyncConfigs) {
                 if (logger.isTraceEnabled()) {
                     logger.trace("Prepared to sync index: {}, destination: {}",
-                        config.getEsMapping().get_index(),
+                        config.getEsMapping().getIndex(),
                         dml.getDestination());
                 }
                 this.sync(config, dml);
                 if (logger.isTraceEnabled()) {
                     logger.trace("Sync completed: {}, destination: {}",
-                        config.getEsMapping().get_index(),
+                        config.getEsMapping().getIndex(),
                         dml.getDestination());
                 }
             }
@@ -73,7 +73,7 @@ public class ESSyncService {
             if (logger.isDebugEnabled()) {
                 StringBuilder configIndexes = new StringBuilder();
                 esSyncConfigs
-                    .forEach(esSyncConfig -> configIndexes.append(esSyncConfig.getEsMapping().get_index()).append(" "));
+                    .forEach(esSyncConfig -> configIndexes.append(esSyncConfig.getEsMapping().getIndex()).append(" "));
                 logger.debug("DML: {} \nAffected indexes: {}",
                     JSON.toJSONString(dml, JSONWriter.Feature.WriteNulls),
                     configIndexes.toString());
@@ -105,10 +105,10 @@ public class ESSyncService {
                 logger.trace("Sync elapsed time: {} ms,destination: {}, es index: {}",
                     (System.currentTimeMillis() - begin),
                     dml.getDestination(),
-                    config.getEsMapping().get_index());
+                    config.getEsMapping().getIndex());
             }
         } catch (Throwable e) {
-            logger.error("sync error, es index: {}, DML : {}", config.getEsMapping().get_index(), dml);
+            logger.error("sync error, es index: {}, DML : {}", config.getEsMapping().getIndex(), dml);
             throw new RuntimeException(e);
         }
     }
@@ -210,7 +210,7 @@ public class ESSyncService {
                 // ------主表 查询sql来更新------
                 if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) {
                     ESMapping mapping = config.getEsMapping();
-                    String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
+                    String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
                     FieldItem idFieldItem = schemaItem.getSelectFields().get(idFieldName);
 
                     boolean idFieldSimple = true;
@@ -336,7 +336,7 @@ public class ESSyncService {
 
             // ------是主表------
             if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) {
-                if (mapping.get_id() != null) {
+                if (mapping.getId() != null) {
                     FieldItem idFieldItem = schemaItem.getIdFieldItem(mapping);
                     // 主键为简单字段
                     if (!idFieldItem.isMethod() && !idFieldItem.isBinaryOp()) {
@@ -349,7 +349,7 @@ public class ESSyncService {
                             logger.trace("Main table delete es index, destination:{}, table: {}, index: {}, id: {}",
                                 config.getDestination(),
                                 dml.getTable(),
-                                mapping.get_index(),
+                                mapping.getIndex(),
                                 idVal);
                         }
                         esTemplate.delete(mapping, idVal, null);
@@ -368,7 +368,7 @@ public class ESSyncService {
                             logger.trace("Main table delete es index, destination:{}, table: {}, index: {}, pk: {}",
                                 config.getDestination(),
                                 dml.getTable(),
-                                mapping.get_index(),
+                                mapping.getIndex(),
                                 pkVal);
                         }
                         esFieldData.remove(pkFieldItem.getFieldName());
@@ -438,7 +438,7 @@ public class ESSyncService {
             logger.trace("Single table insert to es index, destination:{}, table: {}, index: {}, id: {}",
                 config.getDestination(),
                 dml.getTable(),
-                mapping.get_index(),
+                mapping.getIndex(),
                 idVal);
         }
         esTemplate.insert(mapping, idVal, esFieldData);
@@ -461,7 +461,7 @@ public class ESSyncService {
             logger.trace("Main table insert to es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
                 config.getDestination(),
                 dml.getTable(),
-                mapping.get_index(),
+                mapping.getIndex(),
                 sql.replace("\n", " "));
         }
         Util.sqlRS(ds, sql, rs -> {
@@ -475,7 +475,7 @@ public class ESSyncService {
                             "Main table insert to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
                             config.getDestination(),
                             dml.getTable(),
-                            mapping.get_index(),
+                            mapping.getIndex(),
                             idVal);
                     }
                     esTemplate.insert(mapping, idVal, esFieldData);
@@ -497,7 +497,7 @@ public class ESSyncService {
             logger.trace("Main table delete es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
                 config.getDestination(),
                 dml.getTable(),
-                mapping.get_index(),
+                mapping.getIndex(),
                 sql.replace("\n", " "));
         }
         Util.sqlRS(ds, sql, rs -> {
@@ -519,7 +519,7 @@ public class ESSyncService {
                             "Main table delete to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
                             config.getDestination(),
                             dml.getTable(),
-                            mapping.get_index(),
+                            mapping.getIndex(),
                             idVal);
                     }
                     esTemplate.delete(mapping, idVal, esFieldData);
@@ -554,7 +554,7 @@ public class ESSyncService {
 
                     String fieldName = fieldItem.getFieldName();
                     // 判断是否是主键
-                    if (fieldName.equals(mapping.get_id())) {
+                    if (fieldName.equals(mapping.getId())) {
                         fieldName = "_id";
                     }
                     paramsTmp.put(fieldName, value);
@@ -566,7 +566,7 @@ public class ESSyncService {
             logger.trace("Join table update es index by foreign key, destination:{}, table: {}, index: {}",
                 config.getDestination(),
                 dml.getTable(),
-                mapping.get_index());
+                mapping.getIndex());
         }
         esTemplate.updateByQuery(config, paramsTmp, esFieldData);
     }
@@ -617,7 +617,7 @@ public class ESSyncService {
             logger.trace("Join table update es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
                 config.getDestination(),
                 dml.getTable(),
-                mapping.get_index(),
+                mapping.getIndex(),
                 sql.toString().replace("\n", " "));
         }
         Util.sqlRS(ds, sql.toString(), values, rs -> {
@@ -661,7 +661,7 @@ public class ESSyncService {
                                     entry.getKey().getColumn().getColumnName());
                                 String fieldName = fieldItem.getFieldName();
                                 // 判断是否是主键
-                                if (fieldName.equals(mapping.get_id())) {
+                                if (fieldName.equals(mapping.getId())) {
                                     fieldName = "_id";
                                 }
                                 paramsTmp.put(fieldName, value);
@@ -673,7 +673,7 @@ public class ESSyncService {
                         logger.trace("Join table update es index by query sql, destination:{}, table: {}, index: {}",
                             config.getDestination(),
                             dml.getTable(),
-                            mapping.get_index());
+                            mapping.getIndex());
                     }
                     esTemplate.updateByQuery(config, paramsTmp, esFieldData);
                 }
@@ -721,7 +721,7 @@ public class ESSyncService {
             logger.trace("Join table update es index by query whole sql, destination:{}, table: {}, index: {}, sql: {}",
                 config.getDestination(),
                 dml.getTable(),
-                mapping.get_index(),
+                mapping.getIndex(),
                 sql.toString().replace("\n", " "));
         }
         Util.sqlRS(ds, sql.toString(), rs -> {
@@ -775,7 +775,7 @@ public class ESSyncService {
                                 .getValFromRS(mapping, rs, fieldItem.getFieldName(), fieldItem.getFieldName());
                             String fieldName = fieldItem.getFieldName();
                             // 判断是否是主键
-                            if (fieldName.equals(mapping.get_id())) {
+                            if (fieldName.equals(mapping.getId())) {
                                 fieldName = "_id";
                             }
                             paramsTmp.put(fieldName, value);
@@ -787,7 +787,7 @@ public class ESSyncService {
                             "Join table update es index by query whole sql, destination:{}, table: {}, index: {}",
                             config.getDestination(),
                             dml.getTable(),
-                            mapping.get_index());
+                            mapping.getIndex());
                     }
                     esTemplate.updateByQuery(config, paramsTmp, esFieldData);
                 }
@@ -817,7 +817,7 @@ public class ESSyncService {
             logger.trace("Main table update to es index, destination:{}, table: {}, index: {}, id: {}",
                 config.getDestination(),
                 dml.getTable(),
-                mapping.get_index(),
+                mapping.getIndex(),
                 idVal);
         }
         esTemplate.update(mapping, idVal, esFieldData);
@@ -840,7 +840,7 @@ public class ESSyncService {
             logger.trace("Main table update to es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
                 config.getDestination(),
                 dml.getTable(),
-                mapping.get_index(),
+                mapping.getIndex(),
                 sql.replace("\n", " "));
         }
         Util.sqlRS(ds, sql, rs -> {
@@ -854,7 +854,7 @@ public class ESSyncService {
                             "Main table update to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
                             config.getDestination(),
                             dml.getTable(),
-                            mapping.get_index(),
+                            mapping.getIndex(),
                             idVal);
                     }
                     esTemplate.update(mapping, idVal, esFieldData);

client-adapter/hbase/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

client-adapter/kudu/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

client-adapter/launcher/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

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

@@ -59,7 +59,7 @@ esac
 JavaVersion=`$JAVA -version 2>&1 |awk 'NR==1{ gsub(/"/,""); print $3 }' | awk  -F '.' '{print $1}'`
 str=`file -L $JAVA | grep 64-bit`
 
-JAVA_OPTS="$JAVA_OPTS -Xss256k -XX:+AggressiveOpts -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"
+JAVA_OPTS="$JAVA_OPTS -Xss1m -XX:+AggressiveOpts -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"
 if [ $JavaVersion -ge 11 ] ; then
   #JAVA_OPTS="$JAVA_OPTS -Xlog:gc*:$base_log/gc.log:time "
   JAVA_OPTS="$JAVA_OPTS"

client-adapter/logger/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

client-adapter/phoenix/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

client-adapter/pom.xml

@@ -4,12 +4,12 @@
     <parent>
         <groupId>com.alibaba.otter</groupId>
         <artifactId>canal</artifactId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <groupId>com.alibaba.otter</groupId>
     <artifactId>canal.client-adapter</artifactId>
-    <version>1.1.7-SNAPSHOT</version>
+    <version>1.1.7</version>
     <packaging>pom</packaging>
     <name>canal client adapter module for otter ${project.version}</name>
 
@@ -155,7 +155,7 @@
             <dependency>
                 <groupId>org.postgresql</groupId>
                 <artifactId>postgresql</artifactId>
-                <version>42.3.3</version>
+                <version>42.5.1</version>
             </dependency>
             <dependency>
                 <groupId>com.oracle.database.jdbc</groupId>

client-adapter/rdb/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

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

@@ -164,7 +164,8 @@ public class RdbMirrorDbSyncService {
                 logger.trace("Execute DDL sql: {} for database: {}", ddl.getSql(), ddl.getDatabase());
             }
         } catch (Exception e) {
-            throw new RuntimeException(e);
+            logger.error("Execute DDL sql failed sql: {} for database: {} cause: {}",
+                    ddl.getSql(), ddl.getDatabase(), e.getMessage());
         }
     }
 }

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

@@ -1,25 +1,5 @@
 package com.alibaba.otter.canal.client.adapter.rdb.service;
 
-import java.sql.Connection;
-import java.sql.ResultSetMetaData;
-import java.sql.SQLException;
-import java.sql.Types;
-import java.util.ArrayList;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.function.Function;
-
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.fastjson2.JSON;
 import com.alibaba.fastjson2.JSONWriter.Feature;
@@ -30,6 +10,27 @@ import com.alibaba.otter.canal.client.adapter.rdb.support.SingleDml;
 import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil;
 import com.alibaba.otter.canal.client.adapter.support.Dml;
 import com.alibaba.otter.canal.client.adapter.support.Util;
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 /**
  * RDB同步操作业务
@@ -272,7 +273,7 @@ public class RdbSyncService {
         len = insertSql.length();
         insertSql.delete(len - 1, len).append(")");
 
-        Map<String, Integer> ctype = getTargetColumnType(batchExecutor.getConn(), config);
+        Map<String, Integer> ctype = getTargetColumnTypeByColumns(batchExecutor.getConn(), config, columnsMap);
 
         List<Map<String, ?>> values = new ArrayList<>();
         for (Map.Entry<String, String> entry : columnsMap.entrySet()) {
@@ -328,7 +329,7 @@ public class RdbSyncService {
         String backtick = SyncUtil.getBacktickByDbType(dataSource.getDbType());
         Map<String, String> columnsMap = SyncUtil.getColumnsMap(dbMapping, data);
 
-        Map<String, Integer> ctype = getTargetColumnType(batchExecutor.getConn(), config);
+        Map<String, Integer> ctype = getTargetColumnTypeByColumns(batchExecutor.getConn(), config, columnsMap);
 
         StringBuilder updateSql = new StringBuilder();
         updateSql.append("UPDATE ").append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType())).append(" SET ");
@@ -410,6 +411,57 @@ public class RdbSyncService {
         }
     }
 
+    /**
+     * Obtain metadata for the corresponding table columns of the target data source
+     * Check if the metadata of the columns in the src is consistent with the metadata cache of the columns in the cache
+     * (first determine the number of columns and then determine if each column is consistent).
+     * If not, reload
+     * @param conn
+     * @param config
+     * @param columnsMap
+     * @date 2024/6/17 14:45
+     */
+    private Map<String, Integer> getTargetColumnTypeByColumns(Connection conn, MappingConfig config,
+                                                              Map<String, String> columnsMap) {
+        DbMapping dbMapping = config.getDbMapping();
+        String cacheKey = config.getDestination() + "." + dbMapping.getDatabase() + "." + dbMapping.getTable();
+        Map<String, Integer> columnType = getTargetColumnType(conn, config);
+        if(null != columnType && null != columnsMap &&
+                !checkCloumnConsistent(columnType.keySet(), columnsMap.keySet())){
+            logger.info("column meta data not match, reload the column meta data from target db. key:{}", cacheKey);
+            columnsTypeCache.remove(cacheKey);
+            columnType = getTargetColumnType(conn, config);
+            if(null != columnType && null != columnsMap &&
+                    !checkCloumnConsistent(columnType.keySet(), columnsMap.keySet())) {
+                logger.info("the column meta data from db still not match. dbdata:{}", columnType.keySet());
+            }
+        }
+        return columnType;
+    }
+
+
+    /**
+     * 判断binlog的中的列的数据和缓存的列数据是否一致
+     *
+     * @param <T>  集合元素类型
+     * @param column
+     * @param anotherColumn
+     * @return {@link Boolean}
+     */
+    public static<T> Boolean checkCloumnConsistent(Collection<T> column, Collection<T> anotherColumn) {
+        if(null == column || null == anotherColumn || column.size() !=  anotherColumn.size()){
+            return false;
+        }
+        Map<T, T> anotherColumnMap = column.stream().collect(Collectors.toMap(Function.identity(), Function.identity()));
+
+        for (T t : column) {
+            if(null == anotherColumnMap.get(t)){
+                return false;
+            }
+        }
+        return true;
+    }
+
     /**
      * 获取目标字段类型
      *

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

@@ -26,7 +26,7 @@ public class SyncUtil {
     public static Map<String, String> getColumnsMap(MappingConfig.DbMapping dbMapping, Collection<String> columns) {
         Map<String, String> columnsMap;
         if (dbMapping.getMapAll()) {
-            if (dbMapping.getAllMapColumns() != null) {
+            if (dbMapping.getAllMapColumns() != null && columns.size() == dbMapping.getAllMapColumns().size() ) {
                 return dbMapping.getAllMapColumns();
             }
             columnsMap = new LinkedHashMap<>();

client-adapter/tablestore/pom.xml

@@ -3,14 +3,14 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.alibaba.otter</groupId>
     <artifactId>client-adapter.tablestore</artifactId>
     <packaging>jar</packaging>
-    <name>canal client adapter rdb module for otter ${project.version}</name>
+    <name>canal client adapter tablestore module for otter ${project.version}</name>
 
     <dependencies>
         <dependency>

client/pom.xml

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

common/pom.xml

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

connector/core/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.connector</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.alibaba.otter</groupId>

connector/kafka-connector/pom.xml

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

connector/pom.xml

@@ -4,13 +4,13 @@
     <parent>
         <groupId>com.alibaba.otter</groupId>
         <artifactId>canal</artifactId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 
     <groupId>com.alibaba.otter</groupId>
     <artifactId>canal.connector</artifactId>
-    <version>1.1.7-SNAPSHOT</version>
+    <version>1.1.7</version>
     <packaging>pom</packaging>
     <name>canal connector module for otter ${project.version}</name>
 

connector/pulsarmq-connector/pom.xml

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

connector/rabbitmq-connector/pom.xml

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

connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/producer/CanalRabbitMQProducer.java

@@ -72,12 +72,23 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ
         try {
             connect = factory.newConnection();
             channel = connect.createChannel();
-            channel.queueDeclare(rabbitMQProperties.getQueue(), true, false, false, null);
-            channel.exchangeDeclare(rabbitMQProperties
-                .getExchange(), rabbitMQProperties.getDeliveryMode(), true, false, false, null);
-            channel.queueBind(rabbitMQProperties.getQueue(),
-                rabbitMQProperties.getExchange(),
-                rabbitMQProperties.getRoutingKey());
+            String queue=rabbitMQProperties.getQueue();
+            String exchange=rabbitMQProperties.getExchange();
+            String deliveryMode=rabbitMQProperties.getDeliveryMode();
+            String routingKey=rabbitMQProperties.getRoutingKey();
+            if(!StringUtils.isEmpty(queue))
+            {
+                channel.queueDeclare(queue, true, false, false, null);
+            }
+            if(!StringUtils.isEmpty(queue)
+                &&!StringUtils.isEmpty(exchange)
+                &&!StringUtils.isEmpty(deliveryMode)
+                &&!StringUtils.isEmpty(routingKey))
+            {
+                channel.exchangeDeclare(exchange, deliveryMode, true, false, false, null);
+                channel.queueBind(queue,exchange,routingKey);
+            }
+            
 
         } catch (IOException | TimeoutException ex) {
             throw new CanalException("Start RabbitMQ producer error", ex);

connector/rocketmq-connector/pom.xml

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

connector/tcp-connector/pom.xml

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

dbsync/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.1.7-SNAPSHOT</version>
+		<version>1.1.7</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.7-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <groupId>com.alibaba.otter</groupId>

deployer/src/main/bin/startup.sh

@@ -81,7 +81,7 @@ esac
 
 JavaVersion=`$JAVA -version 2>&1 |awk 'NR==1{ gsub(/"/,""); print $3 }' | awk  -F '.' '{print $1}'`
 str=`file -L $JAVA | grep 64-bit`
-JAVA_OPTS="$JAVA_OPTS -Xss256k -XX:+AggressiveOpts -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"
+JAVA_OPTS="$JAVA_OPTS -Xss1m -XX:+AggressiveOpts -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"
 
 if [ $JavaVersion -ge 11 ] ; then
   #JAVA_OPTS="$JAVA_OPTS -Xlog:gc*:$base_log/gc.log:time "

deployer/src/main/resources/canal.properties

@@ -177,6 +177,7 @@ rabbitmq.exchange =
 rabbitmq.username =
 rabbitmq.password =
 rabbitmq.deliveryMode =
+rabbitmq.queue =
 
 
 ##################################################

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

@@ -185,6 +185,9 @@
 		<!--表结构相关-->
 		<property name="enableTsdb" value="${canal.instance.tsdb.enable:true}"/>
 		<property name="tsdbSpringXml" value="${canal.instance.tsdb.spring.xml:}"/>
+		<property name="tsdbJdbcUrl" value="${canal.instance.tsdb.url:}"/>
+		<property name="tsdbJdbcUserName" value="${canal.instance.tsdb.dbUsername:}"/>
+		<property name="tsdbJdbcPassword" value="${canal.instance.tsdb.dbPassword:}"/>
 		<property name="tsdbSnapshotInterval" value="${canal.instance.tsdb.snapshot.interval:24}" />
 		<property name="tsdbSnapshotExpire" value="${canal.instance.tsdb.snapshot.expire:360}" />
 

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

@@ -171,6 +171,9 @@
 		<!--表结构相关-->
 		<property name="enableTsdb" value="${canal.instance.tsdb.enable:true}"/>
 		<property name="tsdbSpringXml" value="${canal.instance.tsdb.spring.xml:}"/>
+		<property name="tsdbJdbcUrl" value="${canal.instance.tsdb.url:}"/>
+		<property name="tsdbJdbcUserName" value="${canal.instance.tsdb.dbUsername:}"/>
+		<property name="tsdbJdbcPassword" value="${canal.instance.tsdb.dbPassword:}"/>
 		<property name="tsdbSnapshotInterval" value="${canal.instance.tsdb.snapshot.interval:24}" />
 		<property name="tsdbSnapshotExpire" value="${canal.instance.tsdb.snapshot.expire:360}" />
 

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

@@ -159,6 +159,9 @@
 		<!--表结构相关-->
 		<property name="enableTsdb" value="${canal.instance.tsdb.enable:false}"/>
 		<property name="tsdbSpringXml" value="${canal.instance.tsdb.spring.xml:}"/>
+		<property name="tsdbJdbcUrl" value="${canal.instance.tsdb.url:}"/>
+		<property name="tsdbJdbcUserName" value="${canal.instance.tsdb.dbUsername:}"/>
+		<property name="tsdbJdbcPassword" value="${canal.instance.tsdb.dbPassword:}"/>
 		<property name="tsdbSnapshotInterval" value="${canal.instance.tsdb.snapshot.interval:24}" />
 		<property name="tsdbSnapshotExpire" value="${canal.instance.tsdb.snapshot.expire:360}" />
 

docker/Dockerfile

@@ -1,4 +1,5 @@
-FROM canal/osbase:v2
+FROM canal/osbase:v3-amd64
+#FROM canal/osbase:v3-arm64
 
 LABEL agapple (jianghang115@gmail.com)
 
@@ -19,7 +20,7 @@ RUN \
     /bin/rm -f /home/admin/canal.deployer-*.tar.gz && \
 
     tar zxvf /tmp/node_exporter.tar.gz -C /home/admin && \
-    ln -s /home/admin/node_exporter-0.18.1.linux-arm64 /home/admin/node_exporter && \
+    ln -s /home/admin/node_exporter-1.6.1* /home/admin/node_exporter && \
 
     mkdir -p home/admin/canal-server/logs  && \
     chmod +x /home/admin/*.sh  && \

docker/Dockerfile_admin

@@ -1,4 +1,5 @@
-FROM canal/osadmin:v1
+FROM canal/osadmin:v3-amd64
+#FROM canal/osadmin:v3-arm64
 
 MAINTAINER agapple (jianghang115@gmail.com)
 

docker/base/Dockerfile_admin

@@ -1,11 +0,0 @@
-FROM canal/osbase:v1
-
-MAINTAINER agapple (jianghang115@gmail.com)
-
-RUN \
-    groupadd -r mysql && useradd -r -g mysql mysql && \
-    yum -y install wget mysql-server --nogpgcheck && \
-    yum clean all && \
-    true
-
-CMD ["/bin/bash"]

docker/base/Dockerfile_v2

@@ -1,11 +0,0 @@
-FROM canal/osbase:v1
-
-MAINTAINER agapple (jianghang115@gmail.com)
-
-env NODE_EPORTER_LINK="https://github.com/prometheus/node_exporter/releases/download/v0.18.1/node_exporter-0.18.1.linux-arm64.tar.gz"
-
-RUN \
-    wget "$NODE_EPORTER_LINK" -O /tmp/node_exporter.tar.gz && \
-    true
-
-CMD ["/bin/bash"]

docker/base/README.md

@@ -0,0 +1,6 @@
+
+# osbase
+cd amd64 && docker build --no-cache -t canal/osbase ./ -f Dockerfile
+
+# osadmin
+cd amd64 && docker build --no-cache -t canal/osadmin ./ -f ./Dockerfile_admin

docker/base/amd64/Dockerfile

@@ -0,0 +1,60 @@
+# amd64
+FROM centos:centos7.9.2009@sha256:dead07b4d8ed7e29e98de0f4504d87e8880d4347859d839686a31da35a3b532f
+
+ARG jdk_rpm
+ARG platform_env
+
+MAINTAINER agapple (jianghang115@gmail.com)
+# env DOWNLOAD_AMD_LINK="https://download.oracle.com/otn/java/jdk/8u361-b09/0ae14417abb444ebb02b9815e2103550/jdk-8u361-linux-x64.rpm"
+# env DOWNLOAD_ARM_LINK="https://download.oracle.com/otn/java/jdk/8u361-b09/0ae14417abb444ebb02b9815e2103550/jdk-8u361-linux-aarch64.rpm"
+env NODE_EPORTER_LINK="https://github.com/prometheus/node_exporter/releases/download/v1.6.1/node_exporter-1.6.1.linux-amd64.tar.gz"
+
+# install system
+# update yum config, fix "centos6.x yum install failure && Determining fastest mirrors slow" problems
+COPY yum/ /tmp/
+RUN \
+    /bin/cp /etc/yum.repos.d/CentOS-Base.repo /etc/yum.repos.d/CentOS-Base.repo.bak && \
+    /bin/cp -f /tmp/CentOS-Base.repo /etc/yum.repos.d/CentOS-Base.repo && \
+    /bin/cp /etc/yum/pluginconf.d/fastestmirror.conf /etc/yum/pluginconf.d/fastestmirror.conf.bak && \
+    awk '{ if($0=="enabled=1"){print "enabled=0";} else{print $0;} }' /etc/yum/pluginconf.d/fastestmirror.conf.bak > /etc/yum/pluginconf.d/fastestmirror.conf && \
+    /bin/cp /etc/yum.conf /etc/yum.conf.bak && \
+    echo 'minrate=1' >> /etc/yum.conf && echo 'timeout=600' >> /etc/yum.conf && \
+    yum clean all && yum makecache
+
+RUN \
+    /bin/cp -f /usr/share/zoneinfo/Asia/Shanghai /etc/localtime && \
+    echo 'root:Hello1234' | chpasswd && \
+    groupadd -r admin && useradd -g admin admin && \
+    yum install -y man && \
+    yum install -y dstat && \
+    yum install -y unzip && \
+    yum install -y nc && \
+    yum install -y openssh-server && \
+    yum install -y tar && \
+    yum install -y which && \
+    yum install -y wget && \
+    yum install -y perl && \
+    yum install -y file && \
+    ssh-keygen -q -N "" -t dsa -f /etc/ssh/ssh_host_dsa_key && \
+    ssh-keygen -q -N "" -t rsa -f /etc/ssh/ssh_host_rsa_key && \
+    sed -ri 's/session    required     pam_loginuid.so/#session    required     pam_loginuid.so/g' /etc/pam.d/sshd && \
+    sed -i -e 's/^#Port 22$/Port 2222/' /etc/ssh/sshd_config && \
+    mkdir -p /root/.ssh && chown root.root /root && chmod 700 /root/.ssh && \
+    yum install -y cronie && \
+    sed -i '/session required pam_loginuid.so/d' /etc/pam.d/crond && \
+    true
+
+COPY ${jdk_rpm} /tmp/
+RUN \
+    touch /var/lib/rpm/* && \
+    #if [ "$BUILDPLATFORM" == "linux/amd64" ] ; then wget --no-cookies --no-check-certificate --header "Cookie: gpw_e24=xxx; oraclelicense=accept-securebackup-cookie" "$DOWNLOAD_AMD_LINK" -O /tmp/jdk-8-linux.rpm  ; fi && \
+    #if [ "$BUILDPLATFORM" == "linux/arm64" ] ; then wget --no-cookies --no-check-certificate --header "Cookie: gpw_e24=xxx; oraclelicense=accept-securebackup-cookie" "$DOWNLOAD_ARM_LINK" -O /tmp/jdk-8-linux.rpm  ; fi && \
+    yum -y install /tmp/jdk-*.rpm && \
+    /bin/rm -f /tmp/jdk-*.rpm && \
+    echo "export JAVA_HOME=/usr/java/latest" >> /etc/profile && \
+    echo "export PATH=\$JAVA_HOME/bin:\$PATH" >> /etc/profile && \
+    wget "$NODE_EPORTER_LINK" -O /tmp/node_exporter.tar.gz && \
+    yum clean all && \
+    true
+
+CMD ["/bin/bash"]

docker/base/amd64/Dockerfile_admin

@@ -0,0 +1,15 @@
+# amd64
+FROM canal/osbase:v3-amd64
+
+MAINTAINER agapple (jianghang115@gmail.com)
+
+RUN \
+    groupadd -r mysql && useradd -r -g mysql mysql && \
+    wget https://dev.mysql.com/get/mysql80-community-release-el7-3.noarch.rpm  && \
+    rpm -ivh mysql80-community-release-el7-3.noarch.rpm  && \
+    yum install sudo mysql-community-server --nogpgcheck -y && \
+    rm -f mysql80-community-release-el7-3.noarch.rpm && \
+    yum clean all && \
+    true
+
+CMD ["/bin/bash"]

docker/base/amd64/yum/CentOS-Base.repo

@@ -0,0 +1,43 @@
+# CentOS-Base.repo
+#
+# The mirror system uses the connecting IP address of the client and the
+# update status of each mirror to pick mirrors that are updated to and
+# geographically close to the client.  You should use this for CentOS updates
+# unless you are manually picking other mirrors.
+#
+# If the mirrorlist= does not work for you, as a fall back you can try the
+# remarked out baseurl= line instead.
+#
+#
+
+[base]
+name=CentOS-7 - Base
+mirrorlist=http://mirrorlist.centos.org/?release=7&arch=$basearch&repo=os&infra=$infra
+#baseurl=http://mirror.centos.org/centos/7/os/$basearch/
+gpgcheck=1
+gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-7
+
+#released updates
+[updates]
+name=CentOS-7 - Updates
+mirrorlist=http://mirrorlist.centos.org/?release=7&arch=$basearch&repo=updates&infra=$infra
+#baseurl=http://mirror.centos.org/centos/7/updates/$basearch/
+gpgcheck=1
+gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-7
+
+#additional packages that may be useful
+[extras]
+name=CentOS-7 - Extras
+mirrorlist=http://mirrorlist.centos.org/?release=7&arch=$basearch&repo=extras&infra=$infra
+#baseurl=http://mirror.centos.org/centos/7/extras/$basearch/
+gpgcheck=1
+gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-7
+
+#additional packages that extend functionality of existing packages
+[centosplus]
+name=CentOS-7 - Plus
+mirrorlist=http://mirrorlist.centos.org/?release=7&arch=$basearch&repo=centosplus&infra=$infra
+#baseurl=http://mirror.centos.org/centos/7/centosplus/$basearch/
+gpgcheck=1
+enabled=0
+gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-7

docker/base/arm64/Dockerfile

@@ -1,10 +1,15 @@
-FROM centos:centos6.10
+# arm64
+FROM centos:centos7.9.2009@sha256:73f11afcbb50d8bc70eab9f0850b3fa30e61a419bc48cf426e63527d14a8373b
+
+ARG jdk_rpm
+ARG platform_env
 
 MAINTAINER agapple (jianghang115@gmail.com)
+# env DOWNLOAD_AMD_LINK="https://download.oracle.com/otn/java/jdk/8u361-b09/0ae14417abb444ebb02b9815e2103550/jdk-8u361-linux-x64.rpm"
+# env DOWNLOAD_ARM_LINK="https://download.oracle.com/otn/java/jdk/8u361-b09/0ae14417abb444ebb02b9815e2103550/jdk-8u361-linux-aarch64.rpm"
+env NODE_EPORTER_LINK="https://github.com/prometheus/node_exporter/releases/download/v1.6.1/node_exporter-1.6.1.linux-arm64.tar.gz"
 
-env DOWNLOAD_LINK="https://download.oracle.com/otn-pub/java/jdk/8u281-b09/89d678f2be164786b292527658ca1605/jdk-8u281-linux-x64.rpm"
 # install system
-
 # update yum config, fix "centos6.x yum install failure && Determining fastest mirrors slow" problems
 COPY yum/ /tmp/
 RUN \
@@ -39,13 +44,14 @@ RUN \
     sed -i '/session required pam_loginuid.so/d' /etc/pam.d/crond && \
     true
 
+COPY ${jdk_rpm} /tmp/
 RUN \
-    touch /var/lib/rpm/* && \ 
-    wget --no-cookies --no-check-certificate --header "Cookie: gpw_e24=xxx; oraclelicense=accept-securebackup-cookie" "$DOWNLOAD_LINK" -O /tmp/jdk-8-linux-x64.rpm && \
-    yum -y install /tmp/jdk-8-linux-x64.rpm && \
-    /bin/rm -f /tmp/jdk-8-linux-x64.rpm && \
+    touch /var/lib/rpm/* && \
+    yum -y install /tmp/jdk-*.rpm && \
+    /bin/rm -f /tmp/jdk-*.rpm && \
     echo "export JAVA_HOME=/usr/java/latest" >> /etc/profile && \
     echo "export PATH=\$JAVA_HOME/bin:\$PATH" >> /etc/profile && \
+    wget "$NODE_EPORTER_LINK" -O /tmp/node_exporter.tar.gz && \
     yum clean all && \
     true
 

docker/base/arm64/Dockerfile_admin

@@ -0,0 +1,15 @@
+# arm64
+FROM canal/osbase:v3-arm64
+
+MAINTAINER agapple (jianghang115@gmail.com)
+
+RUN \
+    groupadd -r mysql && useradd -r -g mysql mysql && \
+    wget https://dev.mysql.com/get/mysql80-community-release-el7-3.noarch.rpm  && \
+    rpm -ivh mysql80-community-release-el7-3.noarch.rpm  && \
+    yum install sudo mysql-community-server-8.0.32 --nogpgcheck -y && \
+    rm -f mysql80-community-release-el7-3.noarch.rpm && \
+    yum clean all && \
+    true
+
+CMD ["/bin/bash"]

docker/base/arm64/yum/CentOS-Base.repo

@@ -0,0 +1,48 @@
+# CentOS-Base.repo
+#
+# The mirror system uses the connecting IP address of the client and the
+# update status of each mirror to pick mirrors that are updated to and
+# geographically close to the client.  You should use this for CentOS updates
+# unless you are manually picking other mirrors.
+#
+# If the mirrorlist= does not work for you, as a fall back you can try the
+# remarked out baseurl= line instead.
+#
+#
+
+[base]
+name=CentOS-$releasever - Base
+mirrorlist=http://mirrorlist.centos.org/?release=$releasever&arch=$basearch&repo=os&infra=$infra
+#baseurl=http://mirror.centos.org/altarch/$releasever/os/$basearch/
+gpgcheck=1
+gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-7
+       file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-7-aarch64
+
+#released updates
+[updates]
+name=CentOS-$releasever - Updates
+mirrorlist=http://mirrorlist.centos.org/?release=$releasever&arch=$basearch&repo=updates&infra=$infra
+#baseurl=http://mirror.centos.org/altarch/$releasever/updates/$basearch/
+gpgcheck=1
+gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-7
+       file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-7-aarch64
+
+#additional packages that may be useful
+[extras]
+name=CentOS-$releasever - Extras
+mirrorlist=http://mirrorlist.centos.org/?release=$releasever&arch=$basearch&repo=extras&infra=$infra
+#baseurl=http://mirror.centos.org/altarch/$releasever/extras/$basearch/
+gpgcheck=1
+gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-7
+       file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-7-aarch64
+enabled=1
+
+#additional packages that extend functionality of existing packages
+[centosplus]
+name=CentOS-$releasever - Plus
+mirrorlist=http://mirrorlist.centos.org/?release=$releasever&arch=$basearch&repo=centosplus&infra=$infra
+#baseurl=http://mirror.centos.org/altarch/$releasever/centosplus/$basearch/
+gpgcheck=1
+enabled=0
+gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-7
+       file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-7-aarch64

docker/base/yum/CentOS-Base.repo

@@ -1,66 +0,0 @@
-# CentOS-Base.repo
-#
-# The mirror system uses the connecting IP address of the client and the
-# update status of each mirror to pick mirrors that are updated to and
-# geographically close to the client.  You should use this for CentOS updates
-# unless you are manually picking other mirrors.
-#
-# If the mirrorlist= does not work for you, as a fall back you can try the
-# remarked out baseurl= line instead.
-#
-
-[base]
-name=CentOS-$releasever - Base
-failovermethod=priority
-#mirrorlist=http://mirrorlist.centos.org/?release=$releasever&arch=$basearch&repo=os&infra=$infra
-baseurl=https://mirrors.aliyun.com/centos-vault/centos/$releasever/os/$basearch/
-        https://mirrors.cloud.tencent.com/centos/$releasever/os/$basearch/
-        https://vault.centos.org/centos/$releasever/os/$basearch/
-gpgcheck=1
-gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-6
-
-#released updates
-[updates]
-name=CentOS-$releasever - Updates
-failovermethod=priority
-#mirrorlist=http://mirrorlist.centos.org/?release=$releasever&arch=$basearch&repo=updates&infra=$infra
-baseurl=https://mirrors.aliyun.com/centos-vault/centos/$releasever/updates/$basearch/
-        https://mirrors.cloud.tencent.com/centos/$releasever/updates/$basearch/
-        https://vault.centos.org/centos/$releasever/updates/$basearch/
-gpgcheck=1
-gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-6
-
-#additional packages that may be useful
-[extras]
-name=CentOS-$releasever - Extras
-failovermethod=priority
-#mirrorlist=http://mirrorlist.centos.org/?release=$releasever&arch=$basearch&repo=extras&infra=$infra
-baseurl=https://mirrors.aliyun.com/centos-vault/centos/$releasever/extras/$basearch/
-        https://mirrors.cloud.tencent.com/centos/$releasever/extras/$basearch/
-        https://vault.centos.org/centos/$releasever/extras/$basearch/
-gpgcheck=1
-gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-6
-
-#additional packages that extend functionality of existing packages
-[centosplus]
-name=CentOS-$releasever - Plus
-failovermethod=priority
-#mirrorlist=http://mirrorlist.centos.org/?release=$releasever&arch=$basearch&repo=centosplus&infra=$infra
-baseurl=https://mirrors.aliyun.com/centos-vault/centos/$releasever/centosplus/$basearch/
-        https://mirrors.cloud.tencent.com/centos/6.0/centosplus/$basearch/
-        https://vault.centos.org/centos/$releasever/centosplus/$basearch/
-gpgcheck=1
-enabled=0
-gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-6
-
-#contrib - packages by Centos Users
-[contrib]
-name=CentOS-$releasever - Contrib
-failovermethod=priority
-#mirrorlist=http://mirrorlist.centos.org/?release=$releasever&arch=$basearch&repo=contrib&infra=$infra
-baseurl=https://mirrors.aliyun.com/centos-vault/centos/$releasever/contrib/$basearch/
-        https://mirrors.cloud.tencent.com/centos/$releasever/contrib/$basearch/
-        https://vault.centos.org/centos/$releasever/contrib/$basearch/
-gpgcheck=1
-enabled=0
-gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-6

docker/build.sh

@@ -14,20 +14,16 @@ case "`uname`" in
 esac
 BASE=${bin_abs_path}
 
-if [ "$1" == "base" ] ; then
-    docker build --no-cache -t canal/osbase $BASE/base
-elif [ "$1" == "base_v2" ] ; then
-    docker build --no-cache -t canal/osbase $BASE/base -f $BASE/base/Dockerfile_v2
-elif [ "$1" == "base_admin" ] ; then
-    docker build --no-cache -t canal/osadmin $BASE/base -f $BASE/base/Dockerfile_admin
-elif [ "$1" == "admin" ] ; then
+if [ "$1" == "admin" ] ; then
     rm -rf $BASE/canal.*.tar.gz ;
     cd $BASE/../ && mvn clean package -Dmaven.test.skip -Denv=release && cd $current_path ;
     cp $BASE/../target/canal.admin-*.tar.gz $BASE/
     docker build --no-cache -t canal/canal-admin $BASE/ -f $BASE/Dockerfile_admin
+    # docker build --platform linux/arm64 --no-cache -t canal/canal-admin $BASE/ -f $BASE/Dockerfile_admin
 else 
     rm -rf $BASE/canal.*.tar.gz ; 
     cd $BASE/../ && mvn clean package -Dmaven.test.skip -Denv=release && cd $current_path ;
     cp $BASE/../target/canal.deployer-*.tar.gz $BASE/
     docker build --no-cache -t canal/canal-server $BASE/
+    # docker build -platform linux/arm64 --no-cache -t canal/canal-server $BASE/
 fi

docker/image/alidata/bin/main.sh

@@ -12,8 +12,8 @@ for e in $(ls /alidata/init/*) ; do
 done
 
 echo "==> INIT DEFAULT"
-service sshd start
-service crond start
+systemctl start sshd
+systemctl start crond
 
 #echo "check hostname -i: `hostname -i`"
 #hti_num=`hostname -i|awk '{print NF}'`
@@ -24,4 +24,4 @@ service crond start
 
 echo "==> INIT DONE"
 echo "==> RUN ${*}"
-exec "${@}"
+exec "${@}"

docker/image/app.sh

@@ -1,5 +1,5 @@
 #!/bin/bash
-set -e
+set +e
 
 source /etc/profile
 export JAVA_HOME=/usr/java/latest
@@ -87,7 +87,7 @@ function start_canal() {
         su admin -c 'cd /home/admin/canal-server/bin/ && sh restart.sh local 1>>/tmp/start.log 2>&1'
         sleep 5
         #check start
-        checkStart "canal" "nc 127.0.0.1 $adminPort -w 1 -z | wc -l" 30
+        checkStart "canal" "nc 127.0.0.1 $adminPort -w 1 -zv 2> /tmp/nc.out && cat /tmp/nc.out | grep -c Connected" 30
     else
         metricsPort=`perl -le 'print $ENV{"canal.metrics.pull.port"}'`
         if [ -z "$metricsPort" ] ; then
@@ -120,7 +120,7 @@ function start_canal() {
         su admin -c 'cd /home/admin/canal-server/bin/ && sh restart.sh 1>>/tmp/start.log 2>&1'
         sleep 5
         #check start
-        checkStart "canal" "nc 127.0.0.1 $metricsPort -w 1 -z | wc -l" 30
+        checkStart "canal" "nc 127.0.0.1 $metricsPort -w 1 -zv 2> /tmp/nc.out && cat /tmp/nc.out | grep -c Connected" 30
     fi  
 }
 

docker/image/app_admin.sh

@@ -1,5 +1,5 @@
 #!/bin/bash
-set -e
+set +e
 
 source /etc/profile
 export JAVA_HOME=/usr/java/latest
@@ -94,37 +94,45 @@ function start_mysql() {
     MYSQL_ROOT_PASSWORD=Hello1234
     # connect local mysql
     if [ -z "$(ls -A /var/lib/mysql)" ]; then
-        mysql_install_db --user=mysql --datadir=/var/lib/mysql 1>>/tmp/start.log 2>&1
-        # These statements _must_ be on individual lines, and _must_ end with
-        # semicolons (no line breaks or comments are permitted).
-        # TODO proper SQL escaping on ALL the things D:
         TEMP_FILE='/tmp/init.sql'
-        echo "update mysql.user set password=password('${MYSQL_ROOT_PASSWORD}') where user='root';" >> $TEMP_FILE
-        echo "grant all privileges on *.* to 'root'@'%' WITH GRANT OPTION ;" >> $TEMP_FILE
+        echo "ALTER USER 'root'@'localhost' IDENTIFIED BY '${MYSQL_ROOT_PASSWORD}';" >> $TEMP_FILE
+        /usr/sbin/mysqld --initialize --user=mysql --datadir=/var/lib/mysql  --init-file=/tmp/init.sql --default-authentication-plugin=mysql_native_password 1>>/tmp/start.log 2>&1
+        echo "default-authentication-plugin=mysql_native_password" >> /etc/my.cnf
+        # systemctl start mysqld
+        sudo -u mysql /usr/sbin/mysqld &
+        sleep 5
+        checkStart "mysql" "echo 'show status' | mysql -b -s -h127.0.0.1 -P3306 -uroot -p${MYSQL_ROOT_PASSWORD} | grep -c Uptime" 30
+        # init file
+        rm -f $TEMP_FILE
         echo "create database if not exists $MYSQL_DATABASE ;" >> $TEMP_FILE
         echo "create user $MYSQL_USER identified by '$MYSQL_USER_PASSWORD' ;" >> $TEMP_FILE
-        echo "grant all privileges on $MYSQL_DATABASE.* to '$MYSQL_USER'@'%' identified by '$MYSQL_USER_PASSWORD' ;" >> $TEMP_FILE
-        echo "grant all privileges on $MYSQL_DATABASE.* to '$MYSQL_USER'@'localhost' identified by '$MYSQL_USER_PASSWORD' ;" >> $TEMP_FILE
+        echo "grant all privileges on $MYSQL_DATABASE.* to '$MYSQL_USER'@'%' ;" >> $TEMP_FILE
         echo "flush privileges;" >> $TEMP_FILE
-        service mysqld start
-        checkStart "mysql" "echo 'show status' | mysql -s -h127.0.0.1 -P3306 -uroot | grep -c Uptime" 30
-        mysql -h127.0.0.1 -uroot -e "source $TEMP_FILE" 1>>/tmp/start.log 2>&1
-
+        # init user
+        cmd="mysql -h127.0.0.1 -uroot -p${MYSQL_ROOT_PASSWORD} -e 'source $TEMP_FILE' 1>>/tmp/start.log 2>&1"
+        eval $cmd
+        /bin/rm -f /tmp/init.sql
+        # init table
         cmd="mysql -h127.0.0.1 -u$MYSQL_USER -p$MYSQL_USER_PASSWORD $MYSQL_DATABASE -e 'source /home/admin/canal-admin/conf/canal_manager.sql' 1>>/tmp/start.log 2>&1"
         eval $cmd
         /bin/rm -f /home/admin/canal-admin/conf/canal_manager.sql
     else
+        echo "recover mysql ..."
         chown -R mysql:mysql /var/lib/mysql
-        service mysqld start
+        # systemctl start mysqld
+        rm -f /var/lib/mysql/mysql.sock.lock
+        sudo -u mysql /usr/sbin/mysqld &
+        sleep 5
         #check start
-        checkStart "mysql" "echo 'show status' | mysql -b -s  -h127.0.0.1 -P3306 -uroot -p$MYSQL_ROOT_PASSWORD | grep -c Uptime" 30
+        checkStart "mysql" "echo 'show status' | mysql -b -s  -h127.0.0.1 -P3306 -uroot -p${MYSQL_ROOT_PASSWORD} | grep -c Uptime" 30
     fi
 }
 
 function stop_mysql() {
     echo "stop mysql ..."
     # stop mysql
-    service mysqld stop
+    # systemctl stop mysqld
+    ps auxf | grep mysqld | grep -v grep | awk '{print $2}' | xargs kill
     echo "stop mysql successful ..."
 }
 
@@ -137,7 +145,7 @@ function start_admin() {
     su admin -c 'cd /home/admin/canal-admin/bin/ && sh restart.sh 1>>/tmp/start.log 2>&1'
     sleep 5
     #check start
-    checkStart "canal" "nc 127.0.0.1 $serverPort -w 1 -z | wc -l" 30
+    checkStart "canal" "nc 127.0.0.1 $serverPort -w 1 -zv 2>/tmp/nc.out && cat /tmp/nc.out | grep -c Connected" 30
 }
 
 function stop_admin() {

docker/run.sh

@@ -97,6 +97,6 @@ fi
 
 MEMORY="-m 4096m"
 LOCALHOST=`getMyIp`
-cmd="docker run -d -it -h $LOCALHOST $CONFIG --name=canal-server $VOLUMNS $NET_MODE $PORTS $MEMORY canal/canal-server"
+cmd="docker run -d --privileged=true -it -h $LOCALHOST $CONFIG --name=canal-server $VOLUMNS $NET_MODE $PORTS $MEMORY canal/canal-server"
 echo $cmd
-eval $cmd
+eval $cmd

docker/run_admin.sh

@@ -86,6 +86,6 @@ fi
 
 MEMORY="-m 1024m"
 LOCALHOST=`getMyIp`
-cmd="docker run -d -it -h $LOCALHOST $CONFIG --name=canal-admin $VOLUMNS $NET_MODE $PORTS $MEMORY canal/canal-admin"
+cmd="docker run -d --privileged=true -it -h $LOCALHOST $CONFIG --name=canal-admin $VOLUMNS $NET_MODE $PORTS $MEMORY canal/canal-admin"
 echo $cmd
 eval $cmd

driver/pom.xml

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

example/pom.xml

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

example/src/main/bin/startup.sh

@@ -61,7 +61,7 @@ esac
 JavaVersion=`$JAVA -version 2>&1 |awk 'NR==1{ gsub(/"/,""); print $3 }' | awk  -F '.' '{print $1}'`
 str=`file -L $JAVA | grep 64-bit`
 
-JAVA_OPTS="$JAVA_OPTS -Xss256k -XX:+AggressiveOpts -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"
+JAVA_OPTS="$JAVA_OPTS -Xss1m -XX:+AggressiveOpts -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"
 if [ $JavaVersion -ge 11 ] ; then
   #JAVA_OPTS="$JAVA_OPTS -Xlog:gc*:$base_log/gc.log:time "
   JAVA_OPTS="$JAVA_OPTS"

filter/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.1.7-SNAPSHOT</version>
+		<version>1.1.7</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.7-SNAPSHOT</version>
+		<version>1.1.7</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.7-SNAPSHOT</version>
+		<version>1.1.7</version>
 		<relativePath>../../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java

@@ -24,13 +24,7 @@ import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
 import com.alibaba.otter.canal.instance.core.AbstractCanalInstance;
 import com.alibaba.otter.canal.instance.manager.model.Canal;
 import com.alibaba.otter.canal.instance.manager.model.CanalParameter;
-import com.alibaba.otter.canal.instance.manager.model.CanalParameter.DataSourcing;
-import com.alibaba.otter.canal.instance.manager.model.CanalParameter.HAMode;
-import com.alibaba.otter.canal.instance.manager.model.CanalParameter.IndexMode;
-import com.alibaba.otter.canal.instance.manager.model.CanalParameter.MetaMode;
-import com.alibaba.otter.canal.instance.manager.model.CanalParameter.SourcingType;
-import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageMode;
-import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageScavengeMode;
+import com.alibaba.otter.canal.instance.manager.model.CanalParameter.*;
 import com.alibaba.otter.canal.meta.FileMixedMetaManager;
 import com.alibaba.otter.canal.meta.MemoryMetaManager;
 import com.alibaba.otter.canal.meta.PeriodMixedMetaManager;
@@ -46,12 +40,7 @@ import com.alibaba.otter.canal.parse.inbound.mysql.rds.RdsBinlogEventParserProxy
 import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DefaultTableMetaTSDBFactory;
 import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDB;
 import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDBBuilder;
-import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
-import com.alibaba.otter.canal.parse.index.FailbackLogPositionManager;
-import com.alibaba.otter.canal.parse.index.MemoryLogPositionManager;
-import com.alibaba.otter.canal.parse.index.MetaLogPositionManager;
-import com.alibaba.otter.canal.parse.index.PeriodMixedLogPositionManager;
-import com.alibaba.otter.canal.parse.index.ZooKeeperLogPositionManager;
+import com.alibaba.otter.canal.parse.index.*;
 import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
 import com.alibaba.otter.canal.protocol.position.EntryPosition;
 import com.alibaba.otter.canal.sink.entry.EntryEventSink;
@@ -338,6 +327,8 @@ public class CanalInstanceWithManager extends AbstractCanalInstance {
                 mysqlEventParser.setTsdbSnapshotExpire(parameters.getTsdbSnapshotExpire());
             }
             boolean tsdbEnable = BooleanUtils.toBoolean(parameters.getTsdbEnable());
+            // manager启动模式默认使用mysql tsdb机制
+            final String tsdbSpringXml = "classpath:spring/tsdb/mysql-tsdb.xml";
             if (tsdbEnable) {
                 mysqlEventParser.setTableMetaTSDBFactory(new DefaultTableMetaTSDBFactory() {
 
@@ -353,9 +344,12 @@ public class CanalInstanceWithManager extends AbstractCanalInstance {
                                 System.setProperty("canal.instance.tsdb.url", parameters.getTsdbJdbcUrl());
                                 System.setProperty("canal.instance.tsdb.dbUsername", parameters.getTsdbJdbcUserName());
                                 System.setProperty("canal.instance.tsdb.dbPassword", parameters.getTsdbJdbcPassword());
+                                System.setProperty("canal.instance.destination", destination);
 
-                                return TableMetaTSDBBuilder.build(destination, "classpath:spring/tsdb/mysql-tsdb.xml");
+                                return TableMetaTSDBBuilder.build(destination, tsdbSpringXml);
                             } finally {
+                                // reset
+                                System.setProperty("canal.instance.destination", "");
                                 System.setProperty("canal.instance.tsdb.url", "");
                                 System.setProperty("canal.instance.tsdb.dbUsername", "");
                                 System.setProperty("canal.instance.tsdb.dbPassword", "");
@@ -363,6 +357,10 @@ public class CanalInstanceWithManager extends AbstractCanalInstance {
                         }
                     }
                 });
+                mysqlEventParser.setTsdbJdbcUrl(parameters.getTsdbJdbcUrl());
+                mysqlEventParser.setTsdbJdbcUserName(parameters.getTsdbJdbcUserName());
+                mysqlEventParser.setTsdbJdbcPassword(parameters.getTsdbJdbcPassword());
+                mysqlEventParser.setTsdbSpringXml(tsdbSpringXml);
                 mysqlEventParser.setEnableTsdb(tsdbEnable);
             }
             eventParser = mysqlEventParser;

instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/PlainCanalInstanceGenerator.java

@@ -61,6 +61,7 @@ public class PlainCanalInstanceGenerator implements CanalInstanceGenerator {
                 throw new CanalException(e);
             } finally {
                 System.setProperty("canal.instance.destination", "");
+                com.alibaba.otter.canal.instance.spring.support.PropertyPlaceholderConfigurer.propertiesLocal.remove();
             }
         }
     }

instance/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.1.7-SNAPSHOT</version>
+		<version>1.1.7</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.7-SNAPSHOT</version>
+		<version>1.1.7</version>
 		<relativePath>../../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

meta/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.1.7-SNAPSHOT</version>
+		<version>1.1.7</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.7-SNAPSHOT</version>
+		<version>1.1.7</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<artifactId>canal.parse</artifactId>

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

@@ -1,6 +1,7 @@
 package com.alibaba.otter.canal.parse.inbound.mysql;
 
 import java.nio.charset.Charset;
+import java.util.Properties;
 import java.util.concurrent.atomic.AtomicLong;
 
 import com.alibaba.otter.canal.filter.CanalEventFilter;
@@ -16,7 +17,6 @@ import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DefaultTableMetaTSDBFact
 import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDB;
 import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDBFactory;
 import com.alibaba.otter.canal.protocol.position.EntryPosition;
-import org.apache.commons.lang.StringUtils;
 
 public abstract class AbstractMysqlEventParser extends AbstractEventParser {
 
@@ -24,6 +24,9 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
 
     protected TableMetaTSDBFactory tableMetaTSDBFactory      = new DefaultTableMetaTSDBFactory();
     protected boolean              enableTsdb                = false;
+    protected String               tsdbJdbcUrl;
+    protected String               tsdbJdbcUserName;
+    protected String               tsdbJdbcPassword;
     protected int                  tsdbSnapshotInterval      = 24;
     protected int                  tsdbSnapshotExpire        = 360;
     protected String               tsdbSpringXml;
@@ -149,14 +152,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
         if (enableTsdb) {
             if (tableMetaTSDB == null) {
                 synchronized (CanalEventParser.class) {
-                    try {
-                        // 设置当前正在加载的通道，加载spring查找文件时会用到该变量
-                        System.setProperty("canal.instance.destination", destination);
-                        // 初始化
-                        tableMetaTSDB = tableMetaTSDBFactory.build(destination, tsdbSpringXml);
-                    } finally {
-                        System.setProperty("canal.instance.destination", "");
-                    }
+                    buildTableMetaTSDB(tsdbSpringXml);
                 }
             }
         }
@@ -173,6 +169,30 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
         super.stop();
     }
 
+    protected synchronized void buildTableMetaTSDB(String tsdbSpringXml) {
+        if (tableMetaTSDB != null) {
+            return;
+        }
+
+        try {
+            // 设置当前正在加载的通道，加载spring查找文件时会用到该变量
+            System.setProperty("canal.instance.tsdb.url", tsdbJdbcUrl);
+            System.setProperty("canal.instance.tsdb.dbUsername", tsdbJdbcUserName);
+            System.setProperty("canal.instance.tsdb.dbPassword", tsdbJdbcPassword);
+            // 初始化
+            this.tableMetaTSDB = tableMetaTSDBFactory.build(destination, tsdbSpringXml);
+        } catch (Throwable e) {
+            logger.warn("failed to build TableMetaTSDB ",e);
+            throw new CanalParseException(e);
+        } finally {
+            // reset
+            Properties props = System.getProperties();
+            props.remove("canal.instance.tsdb.url");
+            props.remove("canal.instance.tsdb.dbUsername");
+            props.remove("canal.instance.tsdb.dbPassword");
+        }
+    }
+
     protected MultiStageCoprocessor buildMultiStageCoprocessor() {
         MysqlMultiStageCoprocessor mysqlMultiStageCoprocessor = new MysqlMultiStageCoprocessor(parallelBufferSize,
             parallelThreadSize,
@@ -255,22 +275,10 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
 
     public void setEnableTsdb(boolean enableTsdb) {
         this.enableTsdb = enableTsdb;
-        if (this.enableTsdb) {
-            if (tableMetaTSDB == null) {
-                // 初始化
-                tableMetaTSDB = tableMetaTSDBFactory.build(destination, tsdbSpringXml);
-            }
-        }
     }
 
     public void setTsdbSpringXml(String tsdbSpringXml) {
         this.tsdbSpringXml = tsdbSpringXml;
-        if (this.enableTsdb) {
-            if (tableMetaTSDB == null) {
-                // 初始化
-                tableMetaTSDB = tableMetaTSDBFactory.build(destination, tsdbSpringXml);
-            }
-        }
     }
 
     public void setTableMetaTSDBFactory(TableMetaTSDBFactory tableMetaTSDBFactory) {
@@ -301,4 +309,15 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
         this.tsdbSnapshotExpire = tsdbSnapshotExpire;
     }
 
+    public void setTsdbJdbcUrl(String tsdbJdbcUrl) {
+        this.tsdbJdbcUrl = tsdbJdbcUrl;
+    }
+
+    public void setTsdbJdbcUserName(String tsdbJdbcUserName) {
+        this.tsdbJdbcUserName = tsdbJdbcUserName;
+    }
+
+    public void setTsdbJdbcPassword(String tsdbJdbcPassword) {
+        this.tsdbJdbcPassword = tsdbJdbcPassword;
+    }
 }

pom.xml

@@ -4,7 +4,7 @@
     <artifactId>canal</artifactId>
     <packaging>pom</packaging>
     <name>canal module for otter ${project.version}</name>
-    <version>1.1.7-SNAPSHOT</version>
+    <version>1.1.7</version>
     <url>https://github.com/alibaba/canal</url>
     <parent>
         <groupId>org.sonatype.oss</groupId>

prometheus/pom.xml

@@ -3,13 +3,13 @@
     <parent>
         <artifactId>canal</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.7</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
     <groupId>com.alibaba.otter</groupId>
     <artifactId>canal.prometheus</artifactId>
-    <version>1.1.7-SNAPSHOT</version>
+    <version>1.1.7</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.7-SNAPSHOT</version>
+		<version>1.1.7</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.7-SNAPSHOT</version>
+		<version>1.1.7</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<artifactId>canal.server</artifactId>

server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java

@@ -73,7 +73,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
         if (!isStart()) {
             super.start();
             // 如果存在provider,则启动metrics service
-            if(metricsPort < 0) {
+            if(metricsPort > 0) {
                 loadCanalMetrics();
                 metrics.setServerPort(metricsPort);
                 metrics.initialize();

sink/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.1.7-SNAPSHOT</version>
+		<version>1.1.7</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.7-SNAPSHOT</version>
+		<version>1.1.7</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>