diff --git a/client-adapter/common/pom.xml b/client-adapter/common/pom.xml index 2bab94c2..15270c11 100644 --- a/client-adapter/common/pom.xml +++ b/client-adapter/common/pom.xml @@ -54,6 +54,11 @@ junit test + + org.yaml + snakeyaml + test + com.diffblue deeptestutils diff --git a/client-adapter/common/src/test/java/com/alibaba/otter/canal/client/adapter/support/YamlUtilsTest.java b/client-adapter/common/src/test/java/com/alibaba/otter/canal/client/adapter/support/YamlUtilsTest.java index d64ffe26..2246c3c1 100644 --- a/client-adapter/common/src/test/java/com/alibaba/otter/canal/client/adapter/support/YamlUtilsTest.java +++ b/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 allMapColumns) { this.allMapColumns = allMapColumns; } + + public String getId() { + return id; + } + + public void setId(String id) { + this.id = id; + } } } diff --git a/client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/ES6xAdapter.java b/client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/ES6xAdapter.java index 85ac4085..647b62b5 100644 --- a/client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/ES6xAdapter.java +++ b/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 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 res = new LinkedHashMap<>(); - res.put("esIndex", mapping.get_index()); + res.put("esIndex", mapping.getIndex()); res.put("count", rowCount); return res; } diff --git a/client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/etl/ESEtlService.java b/client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/etl/ESEtlService.java index 40b4a4ab..b15719e0 100644 --- a/client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/etl/ESEtlService.java +++ b/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 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; diff --git a/client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/support/ES6xTemplate.java b/client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/support/ES6xTemplate.java index 2e81b905..44617dc1 100644 --- a/client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/support/ES6xTemplate.java +++ b/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 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 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 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 dmlOld, Map 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 dmlData, Map 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 dmlData, Map dmlOld, Map 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 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 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<>(); diff --git a/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/ConfigLoadTest.java b/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/ConfigLoadTest.java index 2d5c74b6..a190cba4 100644 --- a/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/ConfigLoadTest.java +++ b/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> dbTableEsSyncConfig = diff --git a/client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/ES7xAdapter.java b/client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/ES7xAdapter.java index 18fd572e..94ad1c17 100644 --- a/client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/ES7xAdapter.java +++ b/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 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 res = new LinkedHashMap<>(); - res.put("esIndex", mapping.get_index()); + res.put("esIndex", mapping.getIndex()); res.put("count", rowCount); return res; } diff --git a/client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/etl/ESEtlService.java b/client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/etl/ESEtlService.java index 05b20cf9..a70ea535 100644 --- a/client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/etl/ESEtlService.java +++ b/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 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; diff --git a/client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/support/ES7xTemplate.java b/client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/support/ES7xTemplate.java index 434cbf9a..4f2ff061 100644 --- a/client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/support/ES7xTemplate.java +++ b/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 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 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 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 dmlOld, Map 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 dmlData, Map 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 dmlData, Map dmlOld, Map 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 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 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<>(); diff --git a/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/ES8xAdapter.java b/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/ES8xAdapter.java index a972fd99..904a44e4 100644 --- a/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/ES8xAdapter.java +++ b/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 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 res = new LinkedHashMap<>(); - res.put("esIndex", mapping.get_index()); + res.put("esIndex", mapping.getIndex()); res.put("count", rowCount); return res; } diff --git a/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/etl/ESEtlService.java b/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/etl/ESEtlService.java index 5c8464c5..3ec64e95 100644 --- a/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/etl/ESEtlService.java +++ b/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 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; diff --git a/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/support/ES8xTemplate.java b/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/support/ES8xTemplate.java index efced987..7f125b7e 100644 --- a/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/support/ES8xTemplate.java +++ b/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 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 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 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 dmlOld, Map 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 dmlData, Map 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 dmlData, Map dmlOld, Map 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 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 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<>(); diff --git a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/ESSyncConfig.java b/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/ESSyncConfig.java index 1277ebbc..e4c31ba3 100644 --- a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/ESSyncConfig.java +++ b/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 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() { diff --git a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/SchemaItem.java b/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/SchemaItem.java index 441e440f..550f474b 100644 --- a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/SchemaItem.java +++ b/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()); } diff --git a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/service/ESSyncService.java b/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/service/ESSyncService.java index e261ec76..4bcb9519 100644 --- a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/service/ESSyncService.java +++ b/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);