Fix Configuration property name '-id' is not valid on canal 1.1.7 alpha 3 alibaba#4879 (alibaba#4887)

* Fix Configuration property name '-id' is not valid

* Fix Configuration property name '-id' is not valid
读取原配置文件

Author: cooperlyt

client-adapter/common/pom.xml

@@ -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 void testLoadConfigToYml() {
             + "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 void testLoadConfigToYml() {
         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 void setDbMapping(DbMapping dbMapping) {
     }
 
     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 Map<String, String> getAllMapColumns() {
         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/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/ES6xAdapter.java

@@ -59,12 +59,12 @@ public void init(OuterAdapterConfig configuration, Properties envProperties) {
     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 ESEtlService(ESConnection esConnection, ESSyncConfig config){
 
     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 @@ protected boolean executeSqlImport(DataSource ds, String sql, List<Object> value
                             }
 
                             // 如果是主键字段则不插入
-                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 @@ protected boolean executeSqlImport(DataSource ds, String sql, List<Object> value
             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 @@ protected boolean executeSqlImport(DataSource ds, String sql, List<Object> value
 
                     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 @@ protected boolean executeSqlImport(DataSource ds, String sql, List<Object> value
                 }
             } 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 @@ protected boolean executeSqlImport(DataSource ds, String sql, List<Object> value
                         (System.currentTimeMillis() - batchBegin),
                         (System.currentTimeMillis() - esBatchBegin),
                         esBulkRequest.numberOfActions(),
-                        mapping.get_index());
+                        mapping.getIndex());
                 }
                 batchBegin = System.currentTimeMillis();
                 esBulkRequest.resetBulk();
@@ -180,12 +180,12 @@ protected boolean executeSqlImport(DataSource ds, String sql, List<Object> value
                     (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;