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 1ffa0c2d..85ac4085 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 @@ -5,6 +5,10 @@ import java.util.List; import java.util.Map; import java.util.Properties; +import javax.sql.DataSource; + +import org.elasticsearch.action.search.SearchResponse; + import com.alibaba.otter.canal.client.adapter.es.core.ESAdapter; import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.es6x.etl.ESEtlService; @@ -14,9 +18,6 @@ import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.EtlResult; import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; import com.alibaba.otter.canal.client.adapter.support.SPI; -import org.elasticsearch.action.search.SearchResponse; - -import javax.sql.DataSource; /** * ES 6.x 外部适配器 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 1f4320b4..40b4a4ab 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 @@ -10,7 +10,6 @@ import java.util.concurrent.atomic.AtomicLong; import javax.sql.DataSource; import org.apache.commons.lang.StringUtils; -import org.elasticsearch.action.bulk.BulkResponse; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.search.SearchHit; @@ -79,119 +78,118 @@ public class ESEtlService extends AbstractEtlService { } // 如果是主键字段则不插入 - if (fieldItem.getFieldName().equals(mapping.get_id())) { - idVal = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); - } else { - Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); - esFieldData.put(Util.cleanColumn(fieldName), val); - } - - } - - if (!mapping.getRelations().isEmpty()) { - mapping.getRelations().forEach((relationField, relationMapping) -> { - Map relations = new HashMap<>(); - relations.put("name", relationMapping.getName()); - if (StringUtils.isNotEmpty(relationMapping.getParent())) { - FieldItem parentFieldItem = mapping.getSchemaItem() - .getSelectFields() - .get(relationMapping.getParent()); - Object parentVal; - try { - parentVal = esTemplate.getValFromRS(mapping, - rs, - parentFieldItem.getFieldName(), - parentFieldItem.getFieldName()); - } catch (SQLException e) { - throw new RuntimeException(e); - } - if (parentVal != null) { - relations.put("parent", parentVal.toString()); - esFieldData.put("$parent_routing", parentVal.toString()); - - } - } - esFieldData.put(Util.cleanColumn(relationField), relations); - }); - } - - 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(), - idVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); - - if (StringUtils.isNotEmpty(parentVal)) { - esUpdateRequest.setRouting(parentVal); - } - - esBulkRequest.add(esUpdateRequest); - } else { - ESIndexRequest esIndexRequest = this.esConnection.new ES6xIndexRequest(mapping - .get_index(), mapping.get_type(), idVal.toString()).setSource(esFieldData); - if (StringUtils.isNotEmpty(parentVal)) { - esIndexRequest.setRouting(parentVal); - } - esBulkRequest.add(esIndexRequest); - } - } 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); - SearchResponse response = esSearchRequest.getResponse(); - for (SearchHit hit : response.getHits()) { - ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping - .get_index(), mapping.get_type(), hit.getId()).setDoc(esFieldData); - esBulkRequest.add(esUpdateRequest); - } - } - - if (esBulkRequest.numberOfActions() % mapping.getCommitBatch() == 0 - && esBulkRequest.numberOfActions() > 0) { - long esBatchBegin = System.currentTimeMillis(); - ESBulkResponse rp = esBulkRequest.bulk(); - if (rp.hasFailures()) { - rp.processFailBulkResponse("全量数据 etl 异常 "); - } - - if (logger.isTraceEnabled()) { - logger.trace("全量数据批量导入批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", - (System.currentTimeMillis() - batchBegin), - (System.currentTimeMillis() - esBatchBegin), - esBulkRequest.numberOfActions(), - mapping.get_index()); - } - batchBegin = System.currentTimeMillis(); - esBulkRequest.resetBulk(); - } - count++; - impCount.incrementAndGet(); - } - - if (esBulkRequest.numberOfActions() > 0) { - long esBatchBegin = System.currentTimeMillis(); - ESBulkResponse rp = esBulkRequest.bulk(); - if (rp.hasFailures()) { - rp.processFailBulkResponse("全量数据 etl 异常 "); - } - if (logger.isTraceEnabled()) { - logger.trace("全量数据批量导入最后批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", - (System.currentTimeMillis() - batchBegin), - (System.currentTimeMillis() - esBatchBegin), - esBulkRequest.numberOfActions(), - mapping.get_index()); - } - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - errMsg.add(mapping.get_index() + " etl failed! ==>" + e.getMessage()); - throw new RuntimeException(e); + if (fieldItem.getFieldName().equals(mapping.get_id())) { + idVal = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); + } else { + Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); + esFieldData.put(Util.cleanColumn(fieldName), val); } - return count; - }); + + } + + if (!mapping.getRelations().isEmpty()) { + mapping.getRelations().forEach((relationField, relationMapping) -> { + Map relations = new HashMap<>(); + relations.put("name", relationMapping.getName()); + if (StringUtils.isNotEmpty(relationMapping.getParent())) { + FieldItem parentFieldItem = mapping.getSchemaItem() + .getSelectFields() + .get(relationMapping.getParent()); + Object parentVal; + try { + parentVal = esTemplate.getValFromRS(mapping, + rs, + parentFieldItem.getFieldName(), + parentFieldItem.getFieldName()); + } catch (SQLException e) { + throw new RuntimeException(e); + } + if (parentVal != null) { + relations.put("parent", parentVal.toString()); + esFieldData.put("$parent_routing", parentVal.toString()); + + } + } + esFieldData.put(Util.cleanColumn(relationField), relations); + }); + } + + 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(), + idVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); + + if (StringUtils.isNotEmpty(parentVal)) { + esUpdateRequest.setRouting(parentVal); + } + + esBulkRequest.add(esUpdateRequest); + } else { + ESIndexRequest esIndexRequest = this.esConnection.new ES6xIndexRequest(mapping.get_index(), + mapping.get_type(), + idVal.toString()).setSource(esFieldData); + if (StringUtils.isNotEmpty(parentVal)) { + esIndexRequest.setRouting(parentVal); + } + esBulkRequest.add(esIndexRequest); + } + } 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); + SearchResponse response = esSearchRequest.getResponse(); + for (SearchHit hit : response.getHits()) { + ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.get_index(), + mapping.get_type(), + hit.getId()).setDoc(esFieldData); + esBulkRequest.add(esUpdateRequest); + } + } + + if (esBulkRequest.numberOfActions() % mapping.getCommitBatch() == 0 && esBulkRequest.numberOfActions() > 0) { + long esBatchBegin = System.currentTimeMillis(); + ESBulkResponse rp = esBulkRequest.bulk(); + if (rp.hasFailures()) { + rp.processFailBulkResponse("全量数据 etl 异常 "); + } + + if (logger.isTraceEnabled()) { + logger.trace("全量数据批量导入批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", + (System.currentTimeMillis() - batchBegin), + (System.currentTimeMillis() - esBatchBegin), + esBulkRequest.numberOfActions(), + mapping.get_index()); + } + batchBegin = System.currentTimeMillis(); + esBulkRequest.resetBulk(); + } + count++; + impCount.incrementAndGet(); + } + + if (esBulkRequest.numberOfActions() > 0) { + long esBatchBegin = System.currentTimeMillis(); + ESBulkResponse rp = esBulkRequest.bulk(); + if (rp.hasFailures()) { + rp.processFailBulkResponse("全量数据 etl 异常 "); + } + if (logger.isTraceEnabled()) { + logger.trace("全量数据批量导入最后批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", + (System.currentTimeMillis() - batchBegin), + (System.currentTimeMillis() - esBatchBegin), + esBulkRequest.numberOfActions(), + mapping.get_index()); + } + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + errMsg.add(mapping.get_index() + " etl failed! ==>" + e.getMessage()); + throw new RuntimeException(e); + } + return count; +} ); return true; } catch (Exception e) { 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 d8df0908..ee74f466 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 @@ -2,7 +2,11 @@ package com.alibaba.otter.canal.client.adapter.es6x.support; import java.sql.ResultSet; import java.sql.SQLException; -import java.util.*; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -20,8 +24,8 @@ import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig.ESMapping; import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem; -import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.FieldItem; import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.ColumnItem; +import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.FieldItem; import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest; import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest.ESBulkResponse; import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest.ESDeleteRequest; diff --git a/client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/support/ESConnection.java b/client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/support/ESConnection.java index 88c34ba6..4870c65f 100644 --- a/client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/support/ESConnection.java +++ b/client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/support/ESConnection.java @@ -5,7 +5,6 @@ import java.net.InetAddress; import java.net.UnknownHostException; import java.util.Map; -import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest; import org.apache.commons.lang.StringUtils; import org.apache.http.HttpHost; import org.apache.http.auth.AuthScope; @@ -27,7 +26,11 @@ import org.elasticsearch.action.search.SearchRequestBuilder; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.action.update.UpdateRequest; import org.elasticsearch.action.update.UpdateRequestBuilder; -import org.elasticsearch.client.*; +import org.elasticsearch.client.RequestOptions; +import org.elasticsearch.client.RestClient; +import org.elasticsearch.client.RestClientBuilder; +import org.elasticsearch.client.RestHighLevelClient; +import org.elasticsearch.client.RestHighLevelClientExt; import org.elasticsearch.client.transport.TransportClient; import org.elasticsearch.cluster.metadata.MappingMetaData; import org.elasticsearch.common.collect.ImmutableOpenMap; @@ -40,6 +43,8 @@ import org.elasticsearch.transport.client.PreBuiltTransportClient; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest; + /** * ES 连接器, Transport Rest 两种方式 * @@ -51,7 +56,7 @@ public class ESConnection { private static final Logger logger = LoggerFactory.getLogger(ESConnection.class); public enum ESClientMode { - TRANSPORT, REST + TRANSPORT, REST } private ESClientMode mode; @@ -86,10 +91,9 @@ public class ESConnection { if (StringUtils.isNotEmpty(nameAndPwd) && nameAndPwd.contains(":")) { String[] nameAndPwdArr = nameAndPwd.split(":"); final CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); - credentialsProvider.setCredentials(AuthScope.ANY, - new UsernamePasswordCredentials(nameAndPwdArr[0], nameAndPwdArr[1])); - restClientBuilder.setHttpClientConfigCallback( - httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)); + credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(nameAndPwdArr[0], + nameAndPwdArr[1])); + restClientBuilder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)); } restHighLevelClient = new RestHighLevelClient(restClientBuilder); } @@ -418,6 +422,7 @@ public class ESConnection { } } + @SuppressWarnings("deprecation") public ESBulkResponse bulk() { if (mode == ESClientMode.TRANSPORT) { BulkResponse responses = bulkRequestBuilder.execute().actionGet(); diff --git a/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/LabelSyncJoinSub2Test.java b/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/LabelSyncJoinSub2Test.java index 21758163..8939e54a 100644 --- a/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/LabelSyncJoinSub2Test.java +++ b/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/LabelSyncJoinSub2Test.java @@ -1,6 +1,10 @@ package com.alibaba.otter.canal.client.adapter.es6x.test.sync; -import java.util.*; +import java.util.ArrayList; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import javax.sql.DataSource; diff --git a/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/LabelSyncJoinSubTest.java b/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/LabelSyncJoinSubTest.java index cf3d4aad..6603086f 100644 --- a/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/LabelSyncJoinSubTest.java +++ b/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/LabelSyncJoinSubTest.java @@ -1,6 +1,10 @@ package com.alibaba.otter.canal.client.adapter.es6x.test.sync; -import java.util.*; +import java.util.ArrayList; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import javax.sql.DataSource; diff --git a/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/RoleSyncJoinOne2Test.java b/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/RoleSyncJoinOne2Test.java index ba87a651..0b8f4bf6 100644 --- a/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/RoleSyncJoinOne2Test.java +++ b/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/RoleSyncJoinOne2Test.java @@ -1,6 +1,10 @@ package com.alibaba.otter.canal.client.adapter.es6x.test.sync; -import java.util.*; +import java.util.ArrayList; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import javax.sql.DataSource; diff --git a/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/RoleSyncJoinOneTest.java b/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/RoleSyncJoinOneTest.java index 5b7eb3bc..37046c29 100644 --- a/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/RoleSyncJoinOneTest.java +++ b/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/RoleSyncJoinOneTest.java @@ -1,6 +1,10 @@ package com.alibaba.otter.canal.client.adapter.es6x.test.sync; -import java.util.*; +import java.util.ArrayList; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import javax.sql.DataSource; diff --git a/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/UserSyncJoinOneTest.java b/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/UserSyncJoinOneTest.java index eb4aa5f3..5cee4bfd 100644 --- a/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/UserSyncJoinOneTest.java +++ b/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/UserSyncJoinOneTest.java @@ -1,6 +1,10 @@ package com.alibaba.otter.canal.client.adapter.es6x.test.sync; -import java.util.*; +import java.util.ArrayList; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import javax.sql.DataSource; diff --git a/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/UserSyncSingleTest.java b/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/UserSyncSingleTest.java index 147bc9f3..43de82e3 100644 --- a/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/UserSyncSingleTest.java +++ b/client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/UserSyncSingleTest.java @@ -1,6 +1,10 @@ package com.alibaba.otter.canal.client.adapter.es6x.test.sync; -import java.util.*; +import java.util.ArrayList; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import org.elasticsearch.action.get.GetResponse; import org.junit.Assert; 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 da4b36ba..18fd572e 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 @@ -5,20 +5,20 @@ import java.util.List; import java.util.Map; import java.util.Properties; -import com.alibaba.otter.canal.client.adapter.es7x.etl.ESEtlService; -import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; +import javax.sql.DataSource; + import org.elasticsearch.action.search.SearchResponse; import com.alibaba.otter.canal.client.adapter.es.core.ESAdapter; import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig; +import com.alibaba.otter.canal.client.adapter.es7x.etl.ESEtlService; import com.alibaba.otter.canal.client.adapter.es7x.support.ES7xTemplate; import com.alibaba.otter.canal.client.adapter.es7x.support.ESConnection; +import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.EtlResult; import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; import com.alibaba.otter.canal.client.adapter.support.SPI; -import javax.sql.DataSource; - /** * ES 7.x 外部适配器 * 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 537b6bef..8dd37761 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 @@ -2,7 +2,11 @@ package com.alibaba.otter.canal.client.adapter.es7x.support; import java.sql.ResultSet; import java.sql.SQLException; -import java.util.*; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; diff --git a/client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/support/ESConnection.java b/client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/support/ESConnection.java index c526721f..082aa64d 100644 --- a/client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/support/ESConnection.java +++ b/client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/support/ESConnection.java @@ -54,11 +54,12 @@ public class ESConnection { private static final Logger logger = LoggerFactory.getLogger(ESConnection.class); public enum ESClientMode { - TRANSPORT, REST + TRANSPORT, REST } private ESClientMode mode; + @SuppressWarnings("deprecation") private TransportClient transportClient; private RestHighLevelClient restHighLevelClient; @@ -89,10 +90,9 @@ public class ESConnection { if (StringUtils.isNotEmpty(nameAndPwd) && nameAndPwd.contains(":")) { String[] nameAndPwdArr = nameAndPwd.split(":"); final CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); - credentialsProvider.setCredentials(AuthScope.ANY, - new UsernamePasswordCredentials(nameAndPwdArr[0], nameAndPwdArr[1])); - restClientBuilder.setHttpClientConfigCallback( - httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)); + credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(nameAndPwdArr[0], + nameAndPwdArr[1])); + restClientBuilder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)); } restHighLevelClient = new RestHighLevelClient(restClientBuilder); } diff --git a/client-adapter/es7x/src/test/java/com/alibaba/otter/canal/client/adapter/es7x/test/ES7xTest.java b/client-adapter/es7x/src/test/java/com/alibaba/otter/canal/client/adapter/es7x/test/ES7xTest.java index 6aa6ade8..a7276d75 100644 --- a/client-adapter/es7x/src/test/java/com/alibaba/otter/canal/client/adapter/es7x/test/ES7xTest.java +++ b/client-adapter/es7x/src/test/java/com/alibaba/otter/canal/client/adapter/es7x/test/ES7xTest.java @@ -24,6 +24,7 @@ import org.junit.Test; @Ignore public class ES7xTest { + @SuppressWarnings("deprecation") private TransportClient transportClient; @Before @@ -65,8 +66,9 @@ public class ES7xTest { relations.put("parent", "2"); BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk(); - bulkRequestBuilder - .add(transportClient.prepareIndex("test", "osm", "2_4").setRouting("2").setSource(esFieldData)); + bulkRequestBuilder.add(transportClient.prepareIndex("test", "osm", "2_4") + .setRouting("2") + .setSource(esFieldData)); commit(bulkRequestBuilder); } diff --git a/client-adapter/es7x/src/test/java/com/alibaba/otter/canal/client/adapter/es7x/test/ESConnectionTest.java b/client-adapter/es7x/src/test/java/com/alibaba/otter/canal/client/adapter/es7x/test/ESConnectionTest.java index 7a36db23..d270a724 100644 --- a/client-adapter/es7x/src/test/java/com/alibaba/otter/canal/client/adapter/es7x/test/ESConnectionTest.java +++ b/client-adapter/es7x/src/test/java/com/alibaba/otter/canal/client/adapter/es7x/test/ESConnectionTest.java @@ -8,9 +8,9 @@ import org.elasticsearch.cluster.metadata.MappingMetaData; import org.junit.Before; import org.junit.Ignore; import org.junit.Test; +import org.springframework.util.Assert; import com.alibaba.otter.canal.client.adapter.es7x.support.ESConnection; -import org.springframework.util.Assert; @Ignore public class ESConnectionTest { diff --git a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/ESAdapter.java b/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/ESAdapter.java index d5bf39dc..56458fc0 100644 --- a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/ESAdapter.java +++ b/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/ESAdapter.java @@ -1,6 +1,5 @@ package com.alibaba.otter.canal.client.adapter.es.core; -import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Properties; @@ -8,7 +7,6 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.regex.Matcher; import java.util.regex.Pattern; -import com.alibaba.otter.canal.client.adapter.es.core.support.ESTemplate; import org.apache.commons.lang.StringUtils; import com.alibaba.druid.pool.DruidDataSource; @@ -19,6 +17,7 @@ import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem; import com.alibaba.otter.canal.client.adapter.es.core.config.SqlParser; import com.alibaba.otter.canal.client.adapter.es.core.monitor.ESConfigMonitor; import com.alibaba.otter.canal.client.adapter.es.core.service.ESSyncService; +import com.alibaba.otter.canal.client.adapter.es.core.support.ESTemplate; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; import com.alibaba.otter.canal.client.adapter.support.EtlResult; @@ -63,8 +62,8 @@ public abstract class ESAdapter implements OuterAdapter { // 过滤不匹配的key的配置 esSyncConfigTmp.forEach((key, config) -> { if ((config.getOuterAdapterKey() == null && configuration.getKey() == null) - || (config.getOuterAdapterKey() != null - && config.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) { + || (config.getOuterAdapterKey() != null && config.getOuterAdapterKey() + .equalsIgnoreCase(configuration.getKey()))) { esSyncConfig.put(key, config); } }); @@ -104,12 +103,12 @@ public abstract class ESAdapter implements OuterAdapter { String table = dml.getTable(); Map configMap; if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) { - configMap = dbTableEsSyncConfig - .get(StringUtils.trimToEmpty(dml.getDestination()) + "-" + StringUtils.trimToEmpty(dml.getGroupId()) - + "_" + database + "-" + table); + configMap = dbTableEsSyncConfig.get(StringUtils.trimToEmpty(dml.getDestination()) + "-" + + StringUtils.trimToEmpty(dml.getGroupId()) + "_" + database + "-" + + table); } else { - configMap = dbTableEsSyncConfig - .get(StringUtils.trimToEmpty(dml.getDestination()) + "_" + database + "-" + table); + configMap = dbTableEsSyncConfig.get(StringUtils.trimToEmpty(dml.getDestination()) + "_" + database + "-" + + table); } if (configMap != null && !configMap.values().isEmpty()) { @@ -155,23 +154,29 @@ public abstract class ESAdapter implements OuterAdapter { } String schema = matcher.group(2); - schemaItem.getAliasTableItems().values().forEach(tableItem -> { - Map esSyncConfigMap; - if (envProperties != null - && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) { - esSyncConfigMap = dbTableEsSyncConfig - .computeIfAbsent(StringUtils.trimToEmpty(config.getDestination()) + "-" - + StringUtils.trimToEmpty(config.getGroupId()) + "_" + schema + "-" - + tableItem.getTableName(), - k -> new ConcurrentHashMap<>()); - } else { - esSyncConfigMap = dbTableEsSyncConfig - .computeIfAbsent(StringUtils.trimToEmpty(config.getDestination()) + "_" + schema + "-" - + tableItem.getTableName(), - k -> new ConcurrentHashMap<>()); - } + schemaItem.getAliasTableItems() + .values() + .forEach(tableItem -> { + Map esSyncConfigMap; + if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) { + esSyncConfigMap = dbTableEsSyncConfig.computeIfAbsent(StringUtils.trimToEmpty(config.getDestination()) + + "-" + + StringUtils.trimToEmpty(config.getGroupId()) + + "_" + + schema + + "-" + + tableItem.getTableName(), + k -> new ConcurrentHashMap<>()); + } else { + esSyncConfigMap = dbTableEsSyncConfig.computeIfAbsent(StringUtils.trimToEmpty(config.getDestination()) + + "_" + + schema + + "-" + + tableItem.getTableName(), + k -> new ConcurrentHashMap<>()); + } - esSyncConfigMap.put(configName, config); - }); + esSyncConfigMap.put(configName, config); + }); } } diff --git a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/SqlParser.java b/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/SqlParser.java index 59d94512..20f445ea 100644 --- a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/SqlParser.java +++ b/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/SqlParser.java @@ -9,8 +9,19 @@ import java.util.stream.Collectors; import com.alibaba.fastsql.sql.SQLUtils; import com.alibaba.fastsql.sql.ast.SQLExpr; -import com.alibaba.fastsql.sql.ast.expr.*; -import com.alibaba.fastsql.sql.ast.statement.*; +import com.alibaba.fastsql.sql.ast.expr.SQLBinaryOpExpr; +import com.alibaba.fastsql.sql.ast.expr.SQLCaseExpr; +import com.alibaba.fastsql.sql.ast.expr.SQLCharExpr; +import com.alibaba.fastsql.sql.ast.expr.SQLIdentifierExpr; +import com.alibaba.fastsql.sql.ast.expr.SQLMethodInvokeExpr; +import com.alibaba.fastsql.sql.ast.expr.SQLPropertyExpr; +import com.alibaba.fastsql.sql.ast.statement.SQLExprTableSource; +import com.alibaba.fastsql.sql.ast.statement.SQLJoinTableSource; +import com.alibaba.fastsql.sql.ast.statement.SQLSelectGroupByClause; +import com.alibaba.fastsql.sql.ast.statement.SQLSelectItem; +import com.alibaba.fastsql.sql.ast.statement.SQLSelectStatement; +import com.alibaba.fastsql.sql.ast.statement.SQLSubqueryTableSource; +import com.alibaba.fastsql.sql.ast.statement.SQLTableSource; import com.alibaba.fastsql.sql.dialect.mysql.ast.statement.MySqlSelectQueryBlock; import com.alibaba.fastsql.sql.dialect.mysql.parser.MySqlStatementParser; import com.alibaba.fastsql.sql.parser.ParserException; diff --git a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/monitor/ESConfigMonitor.java b/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/monitor/ESConfigMonitor.java index 824072e3..2aeea416 100644 --- a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/monitor/ESConfigMonitor.java +++ b/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/monitor/ESConfigMonitor.java @@ -1,13 +1,9 @@ package com.alibaba.otter.canal.client.adapter.es.core.monitor; import java.io.File; -import java.util.HashMap; import java.util.Map; import java.util.Properties; -import java.util.regex.Matcher; -import java.util.regex.Pattern; -import com.alibaba.otter.canal.client.adapter.es.core.ESAdapter; import org.apache.commons.io.filefilter.FileFilterUtils; import org.apache.commons.io.monitor.FileAlterationListenerAdaptor; import org.apache.commons.io.monitor.FileAlterationMonitor; @@ -15,12 +11,9 @@ import org.apache.commons.io.monitor.FileAlterationObserver; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder; +import com.alibaba.otter.canal.client.adapter.es.core.ESAdapter; import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig; -import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem; -import com.alibaba.otter.canal.client.adapter.es.core.config.SqlParser; -import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader; import com.alibaba.otter.canal.client.adapter.support.Util; @@ -70,8 +63,11 @@ public class ESConfigMonitor { try { // 加载新增的配置文件 String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName()); - ESSyncConfig config = YmlConfigBinder - .bindYmlToObj(null, configContent, ESSyncConfig.class, null, envProperties); + ESSyncConfig config = YmlConfigBinder.bindYmlToObj(null, + configContent, + ESSyncConfig.class, + null, + envProperties); if (config != null) { config.validate(); addConfigToCache(file, config); @@ -89,14 +85,17 @@ public class ESConfigMonitor { try { if (esAdapter.getEsSyncConfig().containsKey(file.getName())) { // 加载配置文件 - String configContent = MappingConfigsLoader - .loadConfig(adapterName + File.separator + file.getName()); + String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + + file.getName()); if (configContent == null) { onFileDelete(file); return; } - ESSyncConfig config = YmlConfigBinder - .bindYmlToObj(null, configContent, ESSyncConfig.class, null, envProperties); + ESSyncConfig config = YmlConfigBinder.bindYmlToObj(null, + configContent, + ESSyncConfig.class, + null, + envProperties); if (config == null) { return; } 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 349093bb..4ca23af2 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 @@ -1,22 +1,26 @@ package com.alibaba.otter.canal.client.adapter.es.core.service; -import java.util.*; +import java.util.ArrayList; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import javax.sql.DataSource; -import com.alibaba.fastsql.sql.dialect.mysql.ast.statement.MySqlSelectQueryBlock; -import com.alibaba.otter.canal.client.adapter.es.core.config.SqlParser; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.serializer.SerializerFeature; +import com.alibaba.fastsql.sql.dialect.mysql.ast.statement.MySqlSelectQueryBlock; import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig.ESMapping; import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem; import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.ColumnItem; import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.FieldItem; import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.TableItem; +import com.alibaba.otter.canal.client.adapter.es.core.config.SqlParser; import com.alibaba.otter.canal.client.adapter.es.core.support.ESSyncUtil; import com.alibaba.otter.canal.client.adapter.es.core.support.ESTemplate; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; diff --git a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/support/ESSyncUtil.java b/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/support/ESSyncUtil.java index 3d97a2f0..279f1150 100644 --- a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/support/ESSyncUtil.java +++ b/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/support/ESSyncUtil.java @@ -5,7 +5,13 @@ import java.io.InputStream; import java.nio.charset.StandardCharsets; import java.sql.Blob; import java.sql.SQLException; -import java.util.*; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Date; +import java.util.HashMap; +import java.util.LinkedHashSet; +import java.util.Map; +import java.util.Set; import org.apache.commons.codec.binary.Base64; import org.joda.time.DateTime; diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java index a207be89..d91a49a0 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java @@ -5,19 +5,24 @@ import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Date; import java.util.List; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicLong; import javax.sql.DataSource; -import com.alibaba.otter.canal.client.adapter.support.*; import org.apache.hadoop.hbase.util.Bytes; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig; -import com.alibaba.otter.canal.client.adapter.hbase.support.*; +import com.alibaba.otter.canal.client.adapter.hbase.support.HRow; +import com.alibaba.otter.canal.client.adapter.hbase.support.HbaseTemplate; +import com.alibaba.otter.canal.client.adapter.hbase.support.PhType; +import com.alibaba.otter.canal.client.adapter.hbase.support.PhTypeUtil; +import com.alibaba.otter.canal.client.adapter.hbase.support.Type; +import com.alibaba.otter.canal.client.adapter.hbase.support.TypeUtil; +import com.alibaba.otter.canal.client.adapter.support.AbstractEtlService; +import com.alibaba.otter.canal.client.adapter.support.AdapterConfig; +import com.alibaba.otter.canal.client.adapter.support.EtlResult; +import com.alibaba.otter.canal.client.adapter.support.JdbcTypeUtil; +import com.alibaba.otter.canal.client.adapter.support.Util; import com.google.common.base.Joiner; /** @@ -79,7 +84,8 @@ public class HbaseEtlService extends AbstractEtlService { createTable(); // 拼接sql - String sql = "SELECT * FROM `" + config.getHbaseMapping().getDatabase() + "`.`" + hbaseMapping.getTable() + "`"; + String sql = "SELECT * FROM `" + config.getHbaseMapping().getDatabase() + "`.`" + hbaseMapping.getTable() + + "`"; return super.importData(sql, params); } catch (Exception e) { @@ -120,146 +126,146 @@ public class HbaseEtlService extends AbstractEtlService { if (rowKeyColumns != null) { // 取rowKey字段拼接 - StringBuilder rowKeyVale = new StringBuilder(); - for (String rowKeyColumnName : rowKeyColumns) { - Object obj = rs.getObject(rowKeyColumnName); - if (obj != null) { - rowKeyVale.append(obj.toString()); - } - rowKeyVale.append("|"); - } - int len = rowKeyVale.length(); - if (len > 0) { - rowKeyVale.delete(len - 1, len); - } - row.setRowKey(Bytes.toBytes(rowKeyVale.toString())); + StringBuilder rowKeyVale = new StringBuilder(); + for (String rowKeyColumnName : rowKeyColumns) { + Object obj = rs.getObject(rowKeyColumnName); + if (obj != null) { + rowKeyVale.append(obj.toString()); + } + rowKeyVale.append("|"); + } + int len = rowKeyVale.length(); + if (len > 0) { + rowKeyVale.delete(len - 1, len); + } + row.setRowKey(Bytes.toBytes(rowKeyVale.toString())); + } + + for (int j = 1; j <= cc; j++) { + String columnName = rs.getMetaData().getColumnName(j); + + Object val = JdbcTypeUtil.getRSData(rs, columnName, jdbcTypes[j - 1]); + if (val == null) { + continue; + } + + MappingConfig.ColumnItem columnItem = hbaseMapping.getColumnItems().get(columnName); + // 没有配置映射 + if (columnItem == null) { + String family = hbaseMapping.getFamily(); + String qualifile = columnName; + if (hbaseMapping.isUppercaseQualifier()) { + qualifile = qualifile.toUpperCase(); + } + if (MappingConfig.Mode.STRING == hbaseMapping.getMode()) { + if (hbaseMapping.getRowKey() == null && j == 1) { + row.setRowKey(Bytes.toBytes(val.toString())); + } else { + row.addCell(family, qualifile, Bytes.toBytes(val.toString())); + } + } else if (MappingConfig.Mode.NATIVE == hbaseMapping.getMode()) { + Type type = Type.getType(classes[j - 1]); + if (hbaseMapping.getRowKey() == null && j == 1) { + row.setRowKey(TypeUtil.toBytes(val, type)); + } else { + row.addCell(family, qualifile, TypeUtil.toBytes(val, type)); + } + } else if (MappingConfig.Mode.PHOENIX == hbaseMapping.getMode()) { + PhType phType = PhType.getType(classes[j - 1]); + if (hbaseMapping.getRowKey() == null && j == 1) { + row.setRowKey(PhTypeUtil.toBytes(val, phType)); + } else { + row.addCell(family, qualifile, PhTypeUtil.toBytes(val, phType)); + } + } + } else { + // 如果不需要类型转换 + if (columnItem.getType() == null || "".equals(columnItem.getType())) { + if (val instanceof java.sql.Date) { + SimpleDateFormat dateFmt = new SimpleDateFormat("yyyy-MM-dd"); + val = dateFmt.format((Date) val); + } else if (val instanceof Timestamp) { + SimpleDateFormat datetimeFmt = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); + val = datetimeFmt.format((Date) val); } - for (int j = 1; j <= cc; j++) { - String columnName = rs.getMetaData().getColumnName(j); - - Object val = JdbcTypeUtil.getRSData(rs, columnName, jdbcTypes[j - 1]); - if (val == null) { - continue; + byte[] valBytes = Bytes.toBytes(val.toString()); + if (columnItem.isRowKey()) { + if (columnItem.getRowKeyLen() != null) { + valBytes = Bytes.toBytes(limitLenNum(columnItem.getRowKeyLen(), val)); + row.setRowKey(valBytes); + } else { + row.setRowKey(valBytes); } - - MappingConfig.ColumnItem columnItem = hbaseMapping.getColumnItems().get(columnName); - // 没有配置映射 - if (columnItem == null) { - String family = hbaseMapping.getFamily(); - String qualifile = columnName; - if (hbaseMapping.isUppercaseQualifier()) { - qualifile = qualifile.toUpperCase(); + } else { + row.addCell(columnItem.getFamily(), columnItem.getQualifier(), valBytes); + } + } else { + if (MappingConfig.Mode.STRING == hbaseMapping.getMode()) { + byte[] valBytes = Bytes.toBytes(val.toString()); + if (columnItem.isRowKey()) { + if (columnItem.getRowKeyLen() != null) { + valBytes = Bytes.toBytes(limitLenNum(columnItem.getRowKeyLen(), val)); } - if (MappingConfig.Mode.STRING == hbaseMapping.getMode()) { - if (hbaseMapping.getRowKey() == null && j == 1) { - row.setRowKey(Bytes.toBytes(val.toString())); - } else { - row.addCell(family, qualifile, Bytes.toBytes(val.toString())); - } - } else if (MappingConfig.Mode.NATIVE == hbaseMapping.getMode()) { - Type type = Type.getType(classes[j - 1]); - if (hbaseMapping.getRowKey() == null && j == 1) { - row.setRowKey(TypeUtil.toBytes(val, type)); - } else { - row.addCell(family, qualifile, TypeUtil.toBytes(val, type)); - } - } else if (MappingConfig.Mode.PHOENIX == hbaseMapping.getMode()) { - PhType phType = PhType.getType(classes[j - 1]); - if (hbaseMapping.getRowKey() == null && j == 1) { - row.setRowKey(PhTypeUtil.toBytes(val, phType)); - } else { - row.addCell(family, qualifile, PhTypeUtil.toBytes(val, phType)); - } + row.setRowKey(valBytes); + } else { + row.addCell(columnItem.getFamily(), columnItem.getQualifier(), valBytes); + } + } else if (MappingConfig.Mode.NATIVE == hbaseMapping.getMode()) { + Type type = Type.getType(columnItem.getType()); + if (columnItem.isRowKey()) { + if (columnItem.getRowKeyLen() != null) { + String v = limitLenNum(columnItem.getRowKeyLen(), val); + row.setRowKey(Bytes.toBytes(v)); + } else { + row.setRowKey(TypeUtil.toBytes(val, type)); } } else { - // 如果不需要类型转换 - if (columnItem.getType() == null || "".equals(columnItem.getType())) { - if (val instanceof java.sql.Date) { - SimpleDateFormat dateFmt = new SimpleDateFormat("yyyy-MM-dd"); - val = dateFmt.format((Date) val); - } else if (val instanceof Timestamp) { - SimpleDateFormat datetimeFmt = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); - val = datetimeFmt.format((Date) val); - } - - byte[] valBytes = Bytes.toBytes(val.toString()); - if (columnItem.isRowKey()) { - if (columnItem.getRowKeyLen() != null) { - valBytes = Bytes.toBytes(limitLenNum(columnItem.getRowKeyLen(), val)); - row.setRowKey(valBytes); - } else { - row.setRowKey(valBytes); - } - } else { - row.addCell(columnItem.getFamily(), columnItem.getQualifier(), valBytes); - } - } else { - if (MappingConfig.Mode.STRING == hbaseMapping.getMode()) { - byte[] valBytes = Bytes.toBytes(val.toString()); - if (columnItem.isRowKey()) { - if (columnItem.getRowKeyLen() != null) { - valBytes = Bytes.toBytes(limitLenNum(columnItem.getRowKeyLen(), val)); - } - row.setRowKey(valBytes); - } else { - row.addCell(columnItem.getFamily(), columnItem.getQualifier(), valBytes); - } - } else if (MappingConfig.Mode.NATIVE == hbaseMapping.getMode()) { - Type type = Type.getType(columnItem.getType()); - if (columnItem.isRowKey()) { - if (columnItem.getRowKeyLen() != null) { - String v = limitLenNum(columnItem.getRowKeyLen(), val); - row.setRowKey(Bytes.toBytes(v)); - } else { - row.setRowKey(TypeUtil.toBytes(val, type)); - } - } else { - row.addCell(columnItem.getFamily(), - columnItem.getQualifier(), - TypeUtil.toBytes(val, type)); - } - } else if (MappingConfig.Mode.PHOENIX == hbaseMapping.getMode()) { - PhType phType = PhType.getType(columnItem.getType()); - if (columnItem.isRowKey()) { - row.setRowKey(PhTypeUtil.toBytes(val, phType)); - } else { - row.addCell(columnItem.getFamily(), - columnItem.getQualifier(), - PhTypeUtil.toBytes(val, phType)); - } - } - } + row.addCell(columnItem.getFamily(), + columnItem.getQualifier(), + TypeUtil.toBytes(val, type)); + } + } else if (MappingConfig.Mode.PHOENIX == hbaseMapping.getMode()) { + PhType phType = PhType.getType(columnItem.getType()); + if (columnItem.isRowKey()) { + row.setRowKey(PhTypeUtil.toBytes(val, phType)); + } else { + row.addCell(columnItem.getFamily(), + columnItem.getQualifier(), + PhTypeUtil.toBytes(val, phType)); } } - - if (row.getRowKey() == null) throw new RuntimeException("RowKey 值为空"); - - rows.add(row); - complete = false; - if (i % hbaseMapping.getCommitBatch() == 0 && !rows.isEmpty()) { - hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows); - rows.clear(); - complete = true; - } - i++; - impCount.incrementAndGet(); - if (logger.isDebugEnabled()) { - logger.debug("successful import count:" + impCount.get()); - } } - - if (!complete && !rows.isEmpty()) { - hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows); - } - - } catch (Exception e) { - logger.error(hbaseMapping.getHbaseTable() + " etl failed! ==>" + e.getMessage(), e); - errMsg.add(hbaseMapping.getHbaseTable() + " etl failed! ==>" + e.getMessage()); - // throw new RuntimeException(e); } - return i; - }); + } + + if (row.getRowKey() == null) throw new RuntimeException("RowKey 值为空"); + + rows.add(row); + complete = false; + if (i % hbaseMapping.getCommitBatch() == 0 && !rows.isEmpty()) { + hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows); + rows.clear(); + complete = true; + } + i++; + impCount.incrementAndGet(); + if (logger.isDebugEnabled()) { + logger.debug("successful import count:" + impCount.get()); + } + } + + if (!complete && !rows.isEmpty()) { + hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows); + } + + } catch (Exception e) { + logger.error(hbaseMapping.getHbaseTable() + " etl failed! ==>" + e.getMessage(), e); + errMsg.add(hbaseMapping.getHbaseTable() + " etl failed! ==>" + e.getMessage()); + // throw new RuntimeException(e); + } + return i; +} ); return true; } catch (Exception e) { logger.error(e.getMessage(), e); diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfigHolder.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfigHolder.java index 59e9d2cf..650efdf0 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfigHolder.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfigHolder.java @@ -1,10 +1,10 @@ package com.alibaba.otter.canal.adapter.launcher.config; -import com.alibaba.otter.canal.adapter.launcher.monitor.remote.ConfigItem; - import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import com.alibaba.otter.canal.adapter.launcher.monitor.remote.ConfigItem; + /** * @author L.J.R @ 2019-09-03 * @version 1.1.4 diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AdapterProcessor.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AdapterProcessor.java index 287472b5..416d8c95 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AdapterProcessor.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AdapterProcessor.java @@ -8,20 +8,20 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; -import com.alibaba.otter.canal.adapter.launcher.config.SpringContext; -import com.alibaba.otter.canal.client.adapter.support.Util; -import com.alibaba.otter.canal.connector.core.config.CanalConstants; -import com.alibaba.otter.canal.connector.core.spi.ExtensionLoader; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch; +import com.alibaba.otter.canal.adapter.launcher.config.SpringContext; import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; import com.alibaba.otter.canal.client.adapter.support.MessageUtil; +import com.alibaba.otter.canal.client.adapter.support.Util; +import com.alibaba.otter.canal.connector.core.config.CanalConstants; import com.alibaba.otter.canal.connector.core.consumer.CommonMessage; import com.alibaba.otter.canal.connector.core.spi.CanalMsgConsumer; +import com.alibaba.otter.canal.connector.core.spi.ExtensionLoader; /** * 适配处理器 diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java index 62ec8fa5..20e37414 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java @@ -1,8 +1,5 @@ package com.alibaba.otter.canal.adapter.launcher.loader; -import java.net.InetSocketAddress; -import java.net.SocketAddress; -import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -33,13 +30,13 @@ import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; */ public class CanalAdapterLoader { - private static final Logger logger = LoggerFactory.getLogger(CanalAdapterLoader.class); + private static final Logger logger = LoggerFactory.getLogger(CanalAdapterLoader.class); - private CanalClientConfig canalClientConfig; + private CanalClientConfig canalClientConfig; - private Map canalAdapterProcessors = new HashMap<>(); + private Map canalAdapterProcessors = new HashMap<>(); - private ExtensionLoader loader; + private ExtensionLoader loader; public CanalAdapterLoader(CanalClientConfig canalClientConfig){ this.canalClientConfig = canalClientConfig; @@ -60,123 +57,145 @@ public class CanalAdapterLoader { } canalOuterAdapterGroups.add(canalOuterAdapters); - AdapterProcessor adapterProcessor = canalAdapterProcessors.computeIfAbsent(canalAdapter.getInstance()+"|"+StringUtils.trimToEmpty(group.getGroupId()), - f-> new AdapterProcessor(canalClientConfig,canalAdapter.getInstance(), - group.getGroupId(), canalOuterAdapterGroups)); + AdapterProcessor adapterProcessor = canalAdapterProcessors.computeIfAbsent(canalAdapter.getInstance() + + "|" + + StringUtils.trimToEmpty(group.getGroupId()), + f -> new AdapterProcessor(canalClientConfig, + canalAdapter.getInstance(), + group.getGroupId(), + canalOuterAdapterGroups)); adapterProcessor.start(); - logger.info("Start adapter for canal-client mq topic: {} succeed", - canalAdapter.getInstance() + "-" + group.getGroupId()); + logger.info("Start adapter for canal-client mq topic: {} succeed", canalAdapter.getInstance() + "-" + + group.getGroupId()); } } -// if ("tcp".equalsIgnoreCase(canalClientConfig.getMode())) { -// // 初始化canal-client的适配器 -// for (CanalClientConfig.CanalAdapter canalAdapter : canalClientConfig.getCanalAdapters()) { -// List> canalOuterAdapterGroups = new CopyOnWriteArrayList<>(); -// -// for (CanalClientConfig.Group connectorGroup : canalAdapter.getGroups()) { -// List canalOutConnectors = new CopyOnWriteArrayList<>(); -// for (OuterAdapterConfig c : connectorGroup.getOuterAdapters()) { -// loadAdapter(c, canalOutConnectors); -// } -// canalOuterAdapterGroups.add(canalOutConnectors); -// } -// CanalAdapterWorker worker; -// if (StringUtils.isNotEmpty(canalServerHost)) { -// worker = new CanalAdapterWorker(canalClientConfig, -// canalAdapter.getInstance(), -// canalServerHost, -// zkHosts, -// canalOuterAdapterGroups); -// } else if (zkHosts != null) { -// worker = new CanalAdapterWorker(canalClientConfig, -// canalAdapter.getInstance(), -// zkHosts, -// canalOuterAdapterGroups); -// } else { -// throw new RuntimeException("No canal server connector found"); -// } -// canalWorkers.put(canalAdapter.getInstance(), worker); -// worker.start(); -// logger.info("Start adapter for canal instance: {} succeed", canalAdapter.getInstance()); -// } -// } else if ("kafka".equalsIgnoreCase(canalClientConfig.getMode())) { -// // 初始化canal-client-kafka的适配器 -// for (CanalClientConfig.CanalAdapter canalAdapter : canalClientConfig.getCanalAdapters()) { -// for (CanalClientConfig.Group group : canalAdapter.getGroups()) { -// List> canalOuterAdapterGroups = new CopyOnWriteArrayList<>(); -// List canalOuterAdapters = new CopyOnWriteArrayList<>(); -// for (OuterAdapterConfig config : group.getOuterAdapters()) { -// loadAdapter(config, canalOuterAdapters); -// } -// canalOuterAdapterGroups.add(canalOuterAdapters); -// -// CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(canalClientConfig, -// canalClientConfig.getMqServers(), -// canalAdapter.getInstance(), -// group.getGroupId(), -// canalOuterAdapterGroups, -// canalClientConfig.getFlatMessage()); -// canalMQWorker.put(canalAdapter.getInstance() + "-kafka-" + group.getGroupId(), canalKafkaWorker); -// canalKafkaWorker.start(); -// logger.info("Start adapter for canal-client mq topic: {} succeed", -// canalAdapter.getInstance() + "-" + group.getGroupId()); -// } -// } -// } else if ("rocketMQ".equalsIgnoreCase(canalClientConfig.getMode())) { -// // 初始化canal-client-rocketMQ的适配器 -// for (CanalClientConfig.CanalAdapter canalAdapter : canalClientConfig.getCanalAdapters()) { -// for (CanalClientConfig.Group group : canalAdapter.getGroups()) { -// List> canalOuterAdapterGroups = new CopyOnWriteArrayList<>(); -// List canalOuterAdapters = new CopyOnWriteArrayList<>(); -// for (OuterAdapterConfig config : group.getOuterAdapters()) { -// loadAdapter(config, canalOuterAdapters); -// } -// canalOuterAdapterGroups.add(canalOuterAdapters); -// CanalAdapterRocketMQWorker rocketMQWorker = new CanalAdapterRocketMQWorker(canalClientConfig, -// canalClientConfig.getMqServers(), -// canalAdapter.getInstance(), -// group.getGroupId(), -// canalOuterAdapterGroups, -// canalClientConfig.getAccessKey(), -// canalClientConfig.getSecretKey(), -// canalClientConfig.getFlatMessage(), -// canalClientConfig.isEnableMessageTrace(), -// canalClientConfig.getCustomizedTraceTopic(), -// canalClientConfig.getAccessChannel(), -// canalClientConfig.getNamespace()); -// canalMQWorker.put(canalAdapter.getInstance() + "-rocketmq-" + group.getGroupId(), rocketMQWorker); -// rocketMQWorker.start(); -// -// logger.info("Start adapter for canal-client mq topic: {} succeed", -// canalAdapter.getInstance() + "-" + group.getGroupId()); -// } -// } -// } else if ("rabbitMQ".equalsIgnoreCase(canalClientConfig.getMode())) { -// // 初始化canal-client-rabbitMQ的适配器 -// for (CanalClientConfig.CanalAdapter canalAdapter : canalClientConfig.getCanalAdapters()) { -// for (CanalClientConfig.Group group : canalAdapter.getGroups()) { -// List> canalOuterAdapterGroups = new CopyOnWriteArrayList<>(); -// List canalOuterAdapters = new CopyOnWriteArrayList<>(); -// for (OuterAdapterConfig config : group.getOuterAdapters()) { -// loadAdapter(config, canalOuterAdapters); -// } -// canalOuterAdapterGroups.add(canalOuterAdapters); -// CanalAdapterRabbitMQWorker rabbitMQWork = new CanalAdapterRabbitMQWorker(canalClientConfig, -// canalOuterAdapterGroups, -// canalAdapter.getInstance(), -// group.getGroupId(), -// canalClientConfig.getFlatMessage()); -// canalMQWorker.put(canalAdapter.getInstance() + "-rabbitmq-" + group.getGroupId(), rabbitMQWork); -// rabbitMQWork.start(); -// -// logger.info("Start adapter for canal-client mq topic: {} succeed", -// canalAdapter.getInstance() + "-" + group.getGroupId()); -// } -// } -// // CanalAdapterRabbitMQWork -// } + // if ("tcp".equalsIgnoreCase(canalClientConfig.getMode())) { + // // 初始化canal-client的适配器 + // for (CanalClientConfig.CanalAdapter canalAdapter : + // canalClientConfig.getCanalAdapters()) { + // List> canalOuterAdapterGroups = new + // CopyOnWriteArrayList<>(); + // + // for (CanalClientConfig.Group connectorGroup : + // canalAdapter.getGroups()) { + // List canalOutConnectors = new CopyOnWriteArrayList<>(); + // for (OuterAdapterConfig c : connectorGroup.getOuterAdapters()) { + // loadAdapter(c, canalOutConnectors); + // } + // canalOuterAdapterGroups.add(canalOutConnectors); + // } + // CanalAdapterWorker worker; + // if (StringUtils.isNotEmpty(canalServerHost)) { + // worker = new CanalAdapterWorker(canalClientConfig, + // canalAdapter.getInstance(), + // canalServerHost, + // zkHosts, + // canalOuterAdapterGroups); + // } else if (zkHosts != null) { + // worker = new CanalAdapterWorker(canalClientConfig, + // canalAdapter.getInstance(), + // zkHosts, + // canalOuterAdapterGroups); + // } else { + // throw new RuntimeException("No canal server connector found"); + // } + // canalWorkers.put(canalAdapter.getInstance(), worker); + // worker.start(); + // logger.info("Start adapter for canal instance: {} succeed", + // canalAdapter.getInstance()); + // } + // } else if ("kafka".equalsIgnoreCase(canalClientConfig.getMode())) { + // // 初始化canal-client-kafka的适配器 + // for (CanalClientConfig.CanalAdapter canalAdapter : + // canalClientConfig.getCanalAdapters()) { + // for (CanalClientConfig.Group group : canalAdapter.getGroups()) { + // List> canalOuterAdapterGroups = new + // CopyOnWriteArrayList<>(); + // List canalOuterAdapters = new CopyOnWriteArrayList<>(); + // for (OuterAdapterConfig config : group.getOuterAdapters()) { + // loadAdapter(config, canalOuterAdapters); + // } + // canalOuterAdapterGroups.add(canalOuterAdapters); + // + // CanalAdapterKafkaWorker canalKafkaWorker = new + // CanalAdapterKafkaWorker(canalClientConfig, + // canalClientConfig.getMqServers(), + // canalAdapter.getInstance(), + // group.getGroupId(), + // canalOuterAdapterGroups, + // canalClientConfig.getFlatMessage()); + // canalMQWorker.put(canalAdapter.getInstance() + "-kafka-" + + // group.getGroupId(), canalKafkaWorker); + // canalKafkaWorker.start(); + // logger.info("Start adapter for canal-client mq topic: {} succeed", + // canalAdapter.getInstance() + "-" + group.getGroupId()); + // } + // } + // } else if ("rocketMQ".equalsIgnoreCase(canalClientConfig.getMode())) + // { + // // 初始化canal-client-rocketMQ的适配器 + // for (CanalClientConfig.CanalAdapter canalAdapter : + // canalClientConfig.getCanalAdapters()) { + // for (CanalClientConfig.Group group : canalAdapter.getGroups()) { + // List> canalOuterAdapterGroups = new + // CopyOnWriteArrayList<>(); + // List canalOuterAdapters = new CopyOnWriteArrayList<>(); + // for (OuterAdapterConfig config : group.getOuterAdapters()) { + // loadAdapter(config, canalOuterAdapters); + // } + // canalOuterAdapterGroups.add(canalOuterAdapters); + // CanalAdapterRocketMQWorker rocketMQWorker = new + // CanalAdapterRocketMQWorker(canalClientConfig, + // canalClientConfig.getMqServers(), + // canalAdapter.getInstance(), + // group.getGroupId(), + // canalOuterAdapterGroups, + // canalClientConfig.getAccessKey(), + // canalClientConfig.getSecretKey(), + // canalClientConfig.getFlatMessage(), + // canalClientConfig.isEnableMessageTrace(), + // canalClientConfig.getCustomizedTraceTopic(), + // canalClientConfig.getAccessChannel(), + // canalClientConfig.getNamespace()); + // canalMQWorker.put(canalAdapter.getInstance() + "-rocketmq-" + + // group.getGroupId(), rocketMQWorker); + // rocketMQWorker.start(); + // + // logger.info("Start adapter for canal-client mq topic: {} succeed", + // canalAdapter.getInstance() + "-" + group.getGroupId()); + // } + // } + // } else if ("rabbitMQ".equalsIgnoreCase(canalClientConfig.getMode())) + // { + // // 初始化canal-client-rabbitMQ的适配器 + // for (CanalClientConfig.CanalAdapter canalAdapter : + // canalClientConfig.getCanalAdapters()) { + // for (CanalClientConfig.Group group : canalAdapter.getGroups()) { + // List> canalOuterAdapterGroups = new + // CopyOnWriteArrayList<>(); + // List canalOuterAdapters = new CopyOnWriteArrayList<>(); + // for (OuterAdapterConfig config : group.getOuterAdapters()) { + // loadAdapter(config, canalOuterAdapters); + // } + // canalOuterAdapterGroups.add(canalOuterAdapters); + // CanalAdapterRabbitMQWorker rabbitMQWork = new + // CanalAdapterRabbitMQWorker(canalClientConfig, + // canalOuterAdapterGroups, + // canalAdapter.getInstance(), + // group.getGroupId(), + // canalClientConfig.getFlatMessage()); + // canalMQWorker.put(canalAdapter.getInstance() + "-rabbitmq-" + + // group.getGroupId(), rabbitMQWork); + // rabbitMQWork.start(); + // + // logger.info("Start adapter for canal-client mq topic: {} succeed", + // canalAdapter.getInstance() + "-" + group.getGroupId()); + // } + // } + // // CanalAdapterRabbitMQWork + // } } private void loadAdapter(OuterAdapterConfig config, List canalOutConnectors) { diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/DbRemoteConfigLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/DbRemoteConfigLoader.java index 085ff4d7..0ba09bbf 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/DbRemoteConfigLoader.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/DbRemoteConfigLoader.java @@ -15,14 +15,14 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import com.alibaba.otter.canal.common.utils.CommonUtils; -import com.alibaba.otter.canal.common.utils.NamedThreadFactory; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.otter.canal.adapter.launcher.config.AdapterConfigHolder; +import com.alibaba.otter.canal.common.utils.CommonUtils; +import com.alibaba.otter.canal.common.utils.NamedThreadFactory; import com.google.common.base.Joiner; /** diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteAdapterMonitorImpl.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteAdapterMonitorImpl.java index cb3d2ec9..2f0db822 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteAdapterMonitorImpl.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteAdapterMonitorImpl.java @@ -5,10 +5,11 @@ import java.io.FileOutputStream; import java.io.OutputStreamWriter; import java.nio.charset.StandardCharsets; -import com.alibaba.otter.canal.common.utils.CommonUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.alibaba.otter.canal.common.utils.CommonUtils; + /** * 远程配置监听器实现 * diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java index 1b0276dd..40f7e2f7 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java @@ -3,9 +3,10 @@ package com.alibaba.otter.canal.client.adapter.rdb.config; import java.util.LinkedHashMap; import java.util.Map; -import com.alibaba.otter.canal.client.adapter.support.AdapterConfig; import org.apache.commons.lang.StringUtils; +import com.alibaba.otter.canal.client.adapter.support.AdapterConfig; + /** * RDB表映射配置 * @@ -92,19 +93,21 @@ public class MappingConfig implements AdapterConfig { public static class DbMapping implements AdapterMapping { - private boolean mirrorDb = false; // 是否镜像库 - private String database; // 数据库名或schema名 - private String table; // 表名 - private Map targetPk = new LinkedHashMap<>(); // 目标表主键字段 - private boolean mapAll = false; // 映射所有字段 - private String targetDb; // 目标库名 - private String targetTable; // 目标表名 - private Map targetColumns; // 目标表字段映射 + private boolean mirrorDb = false; // 是否镜像库 + private String database; // 数据库名或schema名 + private String table; // 表名 + private Map targetPk = new LinkedHashMap<>(); // 目标表主键字段 + private boolean mapAll = false; // 映射所有字段 + private String targetDb; // 目标库名 + private String targetTable; // 目标表名 + private Map targetColumns; // 目标表字段映射 - private String etlCondition; // etl条件sql + private boolean caseInsensitive = false; // 目标表不区分大小写，默认是否 - private int readBatch = 5000; - private int commitBatch = 5000; // etl等批量提交大小 + private String etlCondition; // etl条件sql + + private int readBatch = 5000; + private int commitBatch = 5000; // etl等批量提交大小 private Map allMapColumns; @@ -179,6 +182,14 @@ public class MappingConfig implements AdapterConfig { this.targetColumns = targetColumns; } + public boolean isCaseInsensitive() { + return caseInsensitive; + } + + public void setCaseInsensitive(boolean caseInsensitive) { + this.caseInsensitive = caseInsensitive; + } + public String getEtlCondition() { return etlCondition; } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java index 31688229..81254527 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java @@ -1,6 +1,10 @@ package com.alibaba.otter.canal.client.adapter.rdb.service; -import java.sql.*; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java index 9d574fc6..52862337 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java @@ -91,9 +91,10 @@ public class RdbMirrorDbSyncService { if (config == null) { return false; } - + // 是否区分大小写 + boolean caseInsensitive = config.getDbMapping().isCaseInsensitive(); if (config.getConcurrent()) { - List singleDmls = SingleDml.dml2SingleDmls(dml); + List singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive); singleDmls.forEach(singleDml -> { int hash = rdbSyncService.pkHash(config.getDbMapping(), singleDml.getData()); RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml); @@ -101,7 +102,7 @@ public class RdbMirrorDbSyncService { }); } else { int hash = 0; - List singleDmls = SingleDml.dml2SingleDmls(dml); + List singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive); singleDmls.forEach(singleDml -> { RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml); rdbSyncService.getDmlsPartition()[hash].add(syncItem); diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java index 0780c4f0..73392751 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java @@ -112,9 +112,8 @@ public class RdbSyncService { futures.add(executorThreads[i].submit(() -> { try { - dmlsPartition[j].forEach(syncItem -> sync(batchExecutors[j], - syncItem.config, - syncItem.singleDml)); + dmlsPartition[j] + .forEach(syncItem -> sync(batchExecutors[j], syncItem.config, syncItem.singleDml)); dmlsPartition[j].clear(); batchExecutors[j].commit(); return true; @@ -152,49 +151,50 @@ public class RdbSyncService { sync(dmls, dml -> { if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) { // DDL - columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable()); - return false; - } else { - // DML - String destination = StringUtils.trimToEmpty(dml.getDestination()); - String groupId = StringUtils.trimToEmpty(dml.getGroupId()); - String database = dml.getDatabase(); - String table = dml.getTable(); - Map configMap; - if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) { - configMap = mappingConfig.get(destination + "-" + groupId + "_" + database + "-" + table); + columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable()); + return false; } else { - configMap = mappingConfig.get(destination + "_" + database + "-" + table); - } - - if (configMap == null) { - return false; - } - - if (configMap.values().isEmpty()) { - return false; - } - - for (MappingConfig config : configMap.values()) { - if (config.getConcurrent()) { - List singleDmls = SingleDml.dml2SingleDmls(dml); - singleDmls.forEach(singleDml -> { - int hash = pkHash(config.getDbMapping(), singleDml.getData()); - SyncItem syncItem = new SyncItem(config, singleDml); - dmlsPartition[hash].add(syncItem); - }); + // DML + String destination = StringUtils.trimToEmpty(dml.getDestination()); + String groupId = StringUtils.trimToEmpty(dml.getGroupId()); + String database = dml.getDatabase(); + String table = dml.getTable(); + Map configMap; + if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) { + configMap = mappingConfig.get(destination + "-" + groupId + "_" + database + "-" + table); } else { - int hash = 0; - List singleDmls = SingleDml.dml2SingleDmls(dml); - singleDmls.forEach(singleDml -> { - SyncItem syncItem = new SyncItem(config, singleDml); - dmlsPartition[hash].add(syncItem); - }); + configMap = mappingConfig.get(destination + "_" + database + "-" + table); } + + if (configMap == null) { + return false; + } + + if (configMap.values().isEmpty()) { + return false; + } + + for (MappingConfig config : configMap.values()) { + boolean caseInsensitive = config.getDbMapping().isCaseInsensitive(); + if (config.getConcurrent()) { + List singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive); + singleDmls.forEach(singleDml -> { + int hash = pkHash(config.getDbMapping(), singleDml.getData()); + SyncItem syncItem = new SyncItem(config, singleDml); + dmlsPartition[hash].add(syncItem); + }); + } else { + int hash = 0; + List singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive); + singleDmls.forEach(singleDml -> { + SyncItem syncItem = new SyncItem(config, singleDml); + dmlsPartition[hash].add(syncItem); + }); + } + } + return true; } - return true; - } - } ); + }); } /** diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SingleDml.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SingleDml.java index bad12249..d7f9afc7 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SingleDml.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SingleDml.java @@ -4,6 +4,8 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; +import org.springframework.util.LinkedCaseInsensitiveMap; + import com.alibaba.otter.canal.client.adapter.support.Dml; public class SingleDml { @@ -63,7 +65,7 @@ public class SingleDml { this.old = old; } - public static List dml2SingleDmls(Dml dml) { + public static List dml2SingleDmls(Dml dml, boolean caseInsensitive) { List singleDmls = new ArrayList<>(); if (dml.getData() != null) { int size = dml.getData().size(); @@ -73,9 +75,17 @@ public class SingleDml { singleDml.setDatabase(dml.getDatabase()); singleDml.setTable(dml.getTable()); singleDml.setType(dml.getType()); - singleDml.setData(dml.getData().get(i)); + Map data = dml.getData().get(i); + if (caseInsensitive) { + data = toCaseInsensitiveMap(data); + } + singleDml.setData(data); if (dml.getOld() != null) { - singleDml.setOld(dml.getOld().get(i)); + Map oldData = dml.getOld().get(i); + if (caseInsensitive) { + oldData = toCaseInsensitiveMap(oldData); + } + singleDml.setOld(oldData); } singleDmls.add(singleDml); } @@ -89,4 +99,14 @@ public class SingleDml { } return singleDmls; } + + public static List dml2SingleDmls(Dml dml) { + return dml2SingleDmls(dml, false); + } + + private static LinkedCaseInsensitiveMap toCaseInsensitiveMap(Map data) { + LinkedCaseInsensitiveMap map = new LinkedCaseInsensitiveMap(); + map.putAll(data); + return map; + } } diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/DBTest.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/DBTest.java index 43503d7e..dd3960ea 100644 --- a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/DBTest.java +++ b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/DBTest.java @@ -11,6 +11,7 @@ import org.junit.Ignore; import org.junit.Test; import com.alibaba.druid.pool.DruidDataSource; + @Ignore public class DBTest { @@ -39,8 +40,7 @@ public class DBTest { Connection conn = dataSource.getConnection(); conn.setAutoCommit(false); - PreparedStatement pstmt = conn - .prepareStatement("insert into user (id,name,role_id,c_time,test1,test2) values (?,?,?,?,?,?)"); + PreparedStatement pstmt = conn.prepareStatement("insert into user (id,name,role_id,c_time,test1,test2) values (?,?,?,?,?,?)"); java.util.Date now = new java.util.Date(); for (int i = 1; i <= 10000; i++) { @@ -67,14 +67,16 @@ public class DBTest { // ResultSetMetaData rsm = rs.getMetaData(); // int cnt = rsm.getColumnCount(); // for (int i = 1; i <= cnt; i++) { - // System.out.println(rsm.getColumnName(i) + " " + rsm.getColumnType(i)); + // System.out.println(rsm.getColumnName(i) + " " + + // rsm.getColumnType(i)); // } // rs.close(); // stmt.close(); // PreparedStatement pstmt = conn - // .prepareStatement("insert into tb_user (id,name,role_id,c_time,test1,test2) + // .prepareStatement("insert into tb_user + // (id,name,role_id,c_time,test1,test2) // values (?,?,?,?,?,?)"); // pstmt.setBigDecimal(1, new BigDecimal("5")); // pstmt.setString(2, "test"); @@ -91,6 +93,7 @@ public class DBTest { dataSource.close(); } + @SuppressWarnings("unused") private String clob2Str(Clob clob) { String content = ""; try (Reader is = clob.getCharacterStream(); BufferedReader buff = new BufferedReader(is)) { diff --git a/client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java index 4211b9e5..7f3eeb80 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java @@ -1,6 +1,5 @@ package com.alibaba.otter.canal.client; -import java.io.IOException; import java.util.concurrent.TimeUnit; import com.alibaba.otter.canal.protocol.Message; diff --git a/common/src/main/java/com/google/common/collect/MigrateMap.java b/common/src/main/java/com/google/common/collect/MigrateMap.java index 74e444da..402bffaf 100644 --- a/common/src/main/java/com/google/common/collect/MigrateMap.java +++ b/common/src/main/java/com/google/common/collect/MigrateMap.java @@ -1,19 +1,134 @@ package com.google.common.collect; +import java.util.Collection; +import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ExecutionException; import com.google.common.base.Function; +import com.google.common.cache.CacheBuilder; +import com.google.common.cache.CacheLoader; +import com.google.common.cache.LoadingCache; public class MigrateMap { - @SuppressWarnings("deprecation") - public static ConcurrentMap makeComputingMap(MapMaker maker, + public static ConcurrentMap makeComputingMap(CacheBuilder builder, Function computingFunction) { - return maker.makeComputingMap(computingFunction); + final Function function = computingFunction; + LoadingCache computingCache = builder.build(new CacheLoader() { + + @Override + public V load(K key) throws Exception { + return function.apply(key); + } + }); + + return new MigrateConcurrentMap(computingCache); } - @SuppressWarnings("deprecation") public static ConcurrentMap makeComputingMap(Function computingFunction) { - return new MapMaker().makeComputingMap(computingFunction); + return makeComputingMap(CacheBuilder.newBuilder(), computingFunction); + } + + final static class MigrateConcurrentMap implements ConcurrentMap { + + private final LoadingCache computingCache; + + private final ConcurrentMap cacheView; + + MigrateConcurrentMap(LoadingCache computingCache){ + this.computingCache = computingCache; + this.cacheView = computingCache.asMap(); + } + + @Override + public int size() { + return cacheView.size(); + } + + @Override + public boolean isEmpty() { + return cacheView.isEmpty(); + } + + @Override + public boolean containsKey(Object key) { + return cacheView.containsKey(key); + } + + @Override + public boolean containsValue(Object value) { + return cacheView.containsValue(value); + } + + @Override + public V get(Object key) { + try { + return computingCache.get((K) key); + } catch (ExecutionException e) { + throw new RuntimeException(e); + } + } + + @Override + public V put(K key, V value) { + return cacheView.put(key, value); + } + + @Override + public V remove(Object key) { + return cacheView.remove(key); + } + + @Override + public void putAll(Map m) { + cacheView.putAll(m); + } + + @Override + public void clear() { + cacheView.clear(); + } + + @Override + public Set keySet() { + return cacheView.keySet(); + } + + @Override + public Collection values() { + return cacheView.values(); + } + + @Override + public Set> entrySet() { + return cacheView.entrySet(); + } + + @Override + public V putIfAbsent(K key, V value) { + return cacheView.putIfAbsent(key, value); + } + + @Override + public boolean remove(Object key, Object value) { + return cacheView.remove(key, value); + } + + @Override + public boolean replace(K key, V oldValue, V newValue) { + return cacheView.replace(key, oldValue, newValue); + } + + @Override + public V replace(K key, V value) { + return cacheView.replace(key, value); + } + + @Override + public String toString() { + return cacheView.toString(); + } } } diff --git a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/consumer/CommonMessage.java b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/consumer/CommonMessage.java index bd7dd1d3..d1fce6b5 100644 --- a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/consumer/CommonMessage.java +++ b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/consumer/CommonMessage.java @@ -12,21 +12,18 @@ public class CommonMessage implements Serializable { private String table; // 表名 private List pkNames; private Boolean isDdl; - private String type; // 类型: - // INSERT - // UPDATE - // DELETE - // binlog executeTime - private Long es; // 执行耗时 - // dml build timeStamp - private Long ts; // 同步时间 - private String sql; // 执行的sql, - // dml - // sql为空 - private List> data; // 数据列表 - private List> old; // 旧数据列表, - // 用于update, - // size和data的size一一对应 + // 类型:INSERT/UPDATE/DELETE + private String type; + // binlog executeTime, 执行耗时 + private Long es; + // dml build timeStamp, 同步时间 + private Long ts; + // 执行的sql,dml sql为空 + private String sql; + // 数据列表 + private List> data; + // 旧数据列表,用于update,size和data的size一一对应 + private List> old; public String getDatabase() { return database; diff --git a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/filter/PatternUtils.java b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/filter/PatternUtils.java index 8f6228fe..649ba311 100644 --- a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/filter/PatternUtils.java +++ b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/filter/PatternUtils.java @@ -7,13 +7,12 @@ import org.apache.oro.text.regex.Pattern; import org.apache.oro.text.regex.PatternCompiler; import org.apache.oro.text.regex.Perl5Compiler; -import com.google.common.collect.MapMaker; +import com.google.common.cache.CacheBuilder; import com.google.common.collect.MigrateMap; public class PatternUtils { - @SuppressWarnings("deprecation") - private static Map patterns = MigrateMap.makeComputingMap(new MapMaker().softValues(), + private static Map patterns = MigrateMap.makeComputingMap(CacheBuilder.newBuilder().softValues(), pattern -> { try { PatternCompiler pc = new Perl5Compiler(); diff --git a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQMessageUtils.java b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQMessageUtils.java index d64791fe..ad2b4c87 100644 --- a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQMessageUtils.java +++ b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQMessageUtils.java @@ -20,8 +20,8 @@ import com.alibaba.otter.canal.protocol.CanalEntry.RowChange; import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.Message; import com.google.common.base.Function; +import com.google.common.cache.CacheBuilder; import com.google.common.collect.Lists; -import com.google.common.collect.MapMaker; import com.google.common.collect.MigrateMap; import com.google.protobuf.ByteString; import com.google.protobuf.InvalidProtocolBufferException; @@ -34,8 +34,8 @@ import com.google.protobuf.InvalidProtocolBufferException; */ public class MQMessageUtils { - @SuppressWarnings("deprecation") - private static Map> partitionDatas = MigrateMap.makeComputingMap(new MapMaker().softValues(), + private static Map> partitionDatas = MigrateMap.makeComputingMap(CacheBuilder.newBuilder() + .softValues(), pkHashConfigs -> { List datas = Lists.newArrayList(); @@ -73,8 +73,8 @@ public class MQMessageUtils { return datas; }); - @SuppressWarnings("deprecation") - private static Map> dynamicTopicDatas = MigrateMap.makeComputingMap(new MapMaker().softValues(), + private static Map> dynamicTopicDatas = MigrateMap.makeComputingMap(CacheBuilder.newBuilder() + .softValues(), new Function>() { public List apply(String pkHashConfigs) { diff --git a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/spi/ExtensionLoader.java b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/spi/ExtensionLoader.java index 8985b4cb..891a079f 100644 --- a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/spi/ExtensionLoader.java +++ b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/spi/ExtensionLoader.java @@ -409,7 +409,6 @@ public class ExtensionLoader { } } - @SuppressWarnings("unused") private static ClassLoader findClassLoader() { return ExtensionLoader.class.getClassLoader(); } diff --git a/connector/tcp-connector/src/main/java/com/alibaba/otter/canal/connector/tcp/consumer/CanalTCPConsumer.java b/connector/tcp-connector/src/main/java/com/alibaba/otter/canal/connector/tcp/consumer/CanalTCPConsumer.java index 0002f3cd..0046f688 100644 --- a/connector/tcp-connector/src/main/java/com/alibaba/otter/canal/connector/tcp/consumer/CanalTCPConsumer.java +++ b/connector/tcp-connector/src/main/java/com/alibaba/otter/canal/connector/tcp/consumer/CanalTCPConsumer.java @@ -27,13 +27,9 @@ import com.alibaba.otter.canal.protocol.Message; @SPI("tcp") public class CanalTCPConsumer implements CanalMsgConsumer { - private static final String PREFIX_TCP_CONFIG = "canal.tcp."; - - private Long currentBatchId = null; - - private CanalConnector canalConnector; - - private int batchSize = 500; + private Long currentBatchId = null; + private CanalConnector canalConnector; + private int batchSize = 500; @Override public void init(Properties properties, String destination, String groupId) { diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java index 7cb7899b..5639cac1 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java @@ -167,7 +167,7 @@ public class JsonConversion { int type_byte = buffer.getUint8(); int position = buffer.position(); // Then there's the length of the value. - int q_max_bytes = (int) Math.min(len, 5); + int q_max_bytes = (int) Math.min(len - 1, 5); long q_tlen = 0; long q_str_len = 0; long q_n = 0; diff --git a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/JsonConversion_Json_ValueTest.java b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/JsonConversion_Json_ValueTest.java index fb651274..733d102c 100644 --- a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/JsonConversion_Json_ValueTest.java +++ b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/JsonConversion_Json_ValueTest.java @@ -1,54 +1,54 @@ package com.taobao.tddl.dbsync.binlog; -import com.taobao.tddl.dbsync.binlog.JsonConversion.Json_Value; -import com.taobao.tddl.dbsync.binlog.JsonConversion.Json_enum_type; -import com.taobao.tddl.dbsync.binlog.JsonConversion; -import com.taobao.tddl.dbsync.binlog.LogBuffer; import org.junit.Assert; import org.junit.Rule; import org.junit.Test; import org.junit.rules.ExpectedException; import org.junit.rules.Timeout; -import java.lang.reflect.Method; +import com.taobao.tddl.dbsync.binlog.JsonConversion.Json_Value; +import com.taobao.tddl.dbsync.binlog.JsonConversion.Json_enum_type; public class JsonConversion_Json_ValueTest { - @Rule public final ExpectedException thrown = ExpectedException.none(); + @Rule + public final ExpectedException thrown = ExpectedException.none(); - @Rule public final Timeout globalTimeout = new Timeout(10000); + @SuppressWarnings("deprecation") + @Rule + public final Timeout globalTimeout = new Timeout(10000); - /* testedClasses: JsonConversion_Json_Value */ - // Test written by Diffblue Cover. + /* testedClasses: JsonConversion_Json_Value */ + // Test written by Diffblue Cover. - @Test - public void constructorInputNullNotNullOutputVoid() { + @Test + public void constructorInputNullNotNullOutputVoid() { - // Arrange - final Json_enum_type t = null; - final String value = ","; + // Arrange + final Json_enum_type t = null; + final String value = ","; - // Act, creating object to test constructor - final Json_Value objectUnderTest = new Json_Value(t, value); + // Act, creating object to test constructor + final Json_Value objectUnderTest = new Json_Value(t, value); - // Assert side effects - Assert.assertEquals(",", objectUnderTest.m_string_value); - } + // Assert side effects + Assert.assertEquals(",", objectUnderTest.m_string_value); + } - // Test written by Diffblue Cover. - @Test - public void parse_valueInputZeroNotNullPositiveNotNullOutputIllegalArgumentException() { + // Test written by Diffblue Cover. + @Test + public void parse_valueInputZeroNotNullPositiveNotNullOutputIllegalArgumentException() { - // Arrange - final int type = 0; - final LogBuffer buffer = new LogBuffer(); - final long len = 4_294_967_297L; - final String charsetName = "3"; + // Arrange + final int type = 0; + final LogBuffer buffer = new LogBuffer(); + final long len = 4_294_967_297L; + final String charsetName = "3"; - // Act - thrown.expect(IllegalArgumentException.class); - JsonConversion.parse_value(type, buffer, len, charsetName); + // Act + thrown.expect(IllegalArgumentException.class); + JsonConversion.parse_value(type, buffer, len, charsetName); - // Method is not expected to return due to exception thrown - } + // Method is not expected to return due to exception thrown + } } diff --git a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/JsonDiffConversionTest.java b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/JsonDiffConversionTest.java index 8e87797e..8401983b 100644 --- a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/JsonDiffConversionTest.java +++ b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/JsonDiffConversionTest.java @@ -1,214 +1,207 @@ package com.taobao.tddl.dbsync.binlog; -import com.taobao.tddl.dbsync.binlog.JsonDiffConversion; -import com.taobao.tddl.dbsync.binlog.LogBuffer; +import java.lang.reflect.InvocationTargetException; + import org.junit.Assert; import org.junit.Rule; import org.junit.Test; import org.junit.rules.ExpectedException; import org.junit.rules.Timeout; -import java.lang.reflect.Array; -import java.lang.reflect.InvocationTargetException; - public class JsonDiffConversionTest { - @Rule public final ExpectedException thrown = ExpectedException.none(); + @Rule + public final ExpectedException thrown = ExpectedException.none(); - @Rule public final Timeout globalTimeout = new Timeout(10000); + @SuppressWarnings("deprecation") + @Rule + public final Timeout globalTimeout = new Timeout(10000); - /* testedClasses: JsonDiffConversion */ - // Test written by Diffblue Cover. - @Test - public void print_json_diffInputNotNullPositiveNotNullZeroNotNullOutputIllegalArgumentException() - throws InvocationTargetException { + /* testedClasses: JsonDiffConversion */ + // Test written by Diffblue Cover. + @Test + public void print_json_diffInputNotNullPositiveNotNullZeroNotNullOutputIllegalArgumentException() + throws InvocationTargetException { - // Arrange - final LogBuffer buffer = new LogBuffer(); - buffer.position = 28; - buffer.semival = 0; - final byte[] myByteArray = {(byte)3, (byte)3, (byte)67, (byte)67, (byte)67, (byte)67, - (byte)67, (byte)66, (byte)67, (byte)66, (byte)67, (byte)66, - (byte)67, (byte)67, (byte)67, (byte)66, (byte)67, (byte)66, - (byte)67, (byte)66, (byte)67, (byte)67, (byte)67, (byte)66, - (byte)67, (byte)67, (byte)67, (byte)66, (byte)2, (byte)66}; - buffer.buffer = myByteArray; - buffer.limit = -1_000_000_065; - buffer.origin = 1_000_000_096; - final long len = 71L; - final String columnName = "foo"; - final int columnIndex = 0; - final String charsetName = "foo"; - try { + // Arrange + final LogBuffer buffer = new LogBuffer(); + buffer.position = 28; + buffer.semival = 0; + final byte[] myByteArray = { (byte) 3, (byte) 3, (byte) 67, (byte) 67, (byte) 67, (byte) 67, (byte) 67, + (byte) 66, (byte) 67, (byte) 66, (byte) 67, (byte) 66, (byte) 67, (byte) 67, (byte) 67, (byte) 66, + (byte) 67, (byte) 66, (byte) 67, (byte) 66, (byte) 67, (byte) 67, (byte) 67, (byte) 66, (byte) 67, + (byte) 67, (byte) 67, (byte) 66, (byte) 2, (byte) 66 }; + buffer.buffer = myByteArray; + buffer.limit = -1_000_000_065; + buffer.origin = 1_000_000_096; + final long len = 71L; + final String columnName = "foo"; + final int columnIndex = 0; + final String charsetName = "foo"; + try { - // Act - thrown.expect(IllegalArgumentException.class); - JsonDiffConversion.print_json_diff(buffer, len, columnName, columnIndex, charsetName); - } catch (IllegalArgumentException ex) { + // Act + thrown.expect(IllegalArgumentException.class); + JsonDiffConversion.print_json_diff(buffer, len, columnName, columnIndex, charsetName); + } catch (IllegalArgumentException ex) { - // Assert side effects - Assert.assertNotNull(buffer); - Assert.assertEquals(30, buffer.position); - Assert.assertEquals(0, buffer.semival); - Assert.assertArrayEquals(new byte[] {(byte)3, (byte)3, (byte)67, (byte)67, (byte)67, - (byte)67, (byte)67, (byte)66, (byte)67, (byte)66, - (byte)67, (byte)66, (byte)67, (byte)67, (byte)67, - (byte)66, (byte)67, (byte)66, (byte)67, (byte)66, - (byte)67, (byte)67, (byte)67, (byte)66, (byte)67, - (byte)67, (byte)67, (byte)66, (byte)2, (byte)66}, - buffer.buffer); - Assert.assertEquals(-1_000_000_065, buffer.limit); - Assert.assertEquals(1_000_000_096, buffer.origin); - throw ex; + // Assert side effects + Assert.assertNotNull(buffer); + Assert.assertEquals(30, buffer.position); + Assert.assertEquals(0, buffer.semival); + Assert.assertArrayEquals(new byte[] { (byte) 3, (byte) 3, (byte) 67, (byte) 67, (byte) 67, (byte) 67, + (byte) 67, (byte) 66, (byte) 67, (byte) 66, (byte) 67, (byte) 66, (byte) 67, (byte) 67, (byte) 67, + (byte) 66, (byte) 67, (byte) 66, (byte) 67, (byte) 66, (byte) 67, (byte) 67, (byte) 67, (byte) 66, + (byte) 67, (byte) 67, (byte) 67, (byte) 66, (byte) 2, (byte) 66 }, buffer.buffer); + Assert.assertEquals(-1_000_000_065, buffer.limit); + Assert.assertEquals(1_000_000_096, buffer.origin); + throw ex; + } } - } - // Test written by Diffblue Cover. - @Test - public void print_json_diffInputNotNullZeroNotNullZeroNotNullOutputIllegalArgumentException() - throws InvocationTargetException { + // Test written by Diffblue Cover. + @Test + public void print_json_diffInputNotNullZeroNotNullZeroNotNullOutputIllegalArgumentException() + throws InvocationTargetException { - // Arrange - final LogBuffer buffer = new LogBuffer(); - buffer.position = 28; - buffer.semival = 0; - final byte[] myByteArray = { - (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, - (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, - (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)3, (byte)2}; - buffer.buffer = myByteArray; - buffer.limit = 31; - buffer.origin = 0; - final long len = 0L; - final String columnName = "foo"; - final int columnIndex = 0; - final String charsetName = "foo"; - try { + // Arrange + final LogBuffer buffer = new LogBuffer(); + buffer.position = 28; + buffer.semival = 0; + final byte[] myByteArray = { (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, + (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, + (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, + (byte) 3, (byte) 2 }; + buffer.buffer = myByteArray; + buffer.limit = 31; + buffer.origin = 0; + final long len = 0L; + final String columnName = "foo"; + final int columnIndex = 0; + final String charsetName = "foo"; + try { - // Act - thrown.expect(IllegalArgumentException.class); - JsonDiffConversion.print_json_diff(buffer, len, columnName, columnIndex, charsetName); - } catch (IllegalArgumentException ex) { + // Act + thrown.expect(IllegalArgumentException.class); + JsonDiffConversion.print_json_diff(buffer, len, columnName, columnIndex, charsetName); + } catch (IllegalArgumentException ex) { - // Assert side effects - Assert.assertNotNull(buffer); - Assert.assertEquals(29, buffer.position); - Assert.assertEquals(0, buffer.semival); - Assert.assertArrayEquals(new byte[] {(byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, - (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, - (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, - (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, (byte)2, - (byte)2, (byte)2, (byte)2, (byte)2, (byte)3, (byte)2}, - buffer.buffer); - Assert.assertEquals(31, buffer.limit); - Assert.assertEquals(0, buffer.origin); - throw ex; + // Assert side effects + Assert.assertNotNull(buffer); + Assert.assertEquals(29, buffer.position); + Assert.assertEquals(0, buffer.semival); + Assert.assertArrayEquals(new byte[] { (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, + (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, + (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, (byte) 2, + (byte) 2, (byte) 3, (byte) 2 }, buffer.buffer); + Assert.assertEquals(31, buffer.limit); + Assert.assertEquals(0, buffer.origin); + throw ex; + } } - } - // Test written by Diffblue Cover. - @Test - public void print_json_diffInputNotNullZeroNotNullZeroNotNullOutputIllegalArgumentException2() - throws InvocationTargetException { + // Test written by Diffblue Cover. + @Test + public void print_json_diffInputNotNullZeroNotNullZeroNotNullOutputIllegalArgumentException2() + throws InvocationTargetException { - // Arrange - final LogBuffer buffer = new LogBuffer(); - buffer.position = 15; - buffer.semival = 0; - final byte[] myByteArray = { - (byte)1, (byte)1, (byte)0, (byte)0, (byte)1, (byte)1, (byte)1, (byte)1, (byte)1, (byte)1, - (byte)0, (byte)1, (byte)1, (byte)1, (byte)1, (byte)0, (byte)1, (byte)1, (byte)0, (byte)0, - (byte)1, (byte)1, (byte)1, (byte)1, (byte)1, (byte)1, (byte)1, (byte)0, (byte)1, (byte)1}; - buffer.buffer = myByteArray; - buffer.limit = -1_215_751_986; - buffer.origin = 1_215_752_002; - final long len = 0L; - final String columnName = "foo"; - final int columnIndex = 0; - final String charsetName = "foo"; - try { + // Arrange + final LogBuffer buffer = new LogBuffer(); + buffer.position = 15; + buffer.semival = 0; + final byte[] myByteArray = { (byte) 1, (byte) 1, (byte) 0, (byte) 0, (byte) 1, (byte) 1, (byte) 1, (byte) 1, + (byte) 1, (byte) 1, (byte) 0, (byte) 1, (byte) 1, (byte) 1, (byte) 1, (byte) 0, (byte) 1, (byte) 1, + (byte) 0, (byte) 0, (byte) 1, (byte) 1, (byte) 1, (byte) 1, (byte) 1, (byte) 1, (byte) 1, (byte) 0, + (byte) 1, (byte) 1 }; + buffer.buffer = myByteArray; + buffer.limit = -1_215_751_986; + buffer.origin = 1_215_752_002; + final long len = 0L; + final String columnName = "foo"; + final int columnIndex = 0; + final String charsetName = "foo"; + try { - // Act - thrown.expect(IllegalArgumentException.class); - JsonDiffConversion.print_json_diff(buffer, len, columnName, columnIndex, charsetName); - } catch (IllegalArgumentException ex) { + // Act + thrown.expect(IllegalArgumentException.class); + JsonDiffConversion.print_json_diff(buffer, len, columnName, columnIndex, charsetName); + } catch (IllegalArgumentException ex) { - // Assert side effects - Assert.assertNotNull(buffer); - Assert.assertEquals(16, buffer.position); - Assert.assertEquals(0, buffer.semival); - Assert.assertArrayEquals(new byte[] {(byte)1, (byte)1, (byte)0, (byte)0, (byte)1, (byte)1, - (byte)1, (byte)1, (byte)1, (byte)1, (byte)0, (byte)1, - (byte)1, (byte)1, (byte)1, (byte)0, (byte)1, (byte)1, - (byte)0, (byte)0, (byte)1, (byte)1, (byte)1, (byte)1, - (byte)1, (byte)1, (byte)1, (byte)0, (byte)1, (byte)1}, - buffer.buffer); - Assert.assertEquals(-1_215_751_986, buffer.limit); - Assert.assertEquals(1_215_752_002, buffer.origin); - throw ex; + // Assert side effects + Assert.assertNotNull(buffer); + Assert.assertEquals(16, buffer.position); + Assert.assertEquals(0, buffer.semival); + Assert.assertArrayEquals(new byte[] { (byte) 1, (byte) 1, (byte) 0, (byte) 0, (byte) 1, (byte) 1, (byte) 1, + (byte) 1, (byte) 1, (byte) 1, (byte) 0, (byte) 1, (byte) 1, (byte) 1, (byte) 1, (byte) 0, (byte) 1, + (byte) 1, (byte) 0, (byte) 0, (byte) 1, (byte) 1, (byte) 1, (byte) 1, (byte) 1, (byte) 1, (byte) 1, + (byte) 0, (byte) 1, (byte) 1 }, buffer.buffer); + Assert.assertEquals(-1_215_751_986, buffer.limit); + Assert.assertEquals(1_215_752_002, buffer.origin); + throw ex; + } } - } - // Test written by Diffblue Cover. - @Test - public void print_json_diffInputNotNullZeroNotNullZeroNotNullOutputIllegalArgumentException3() - throws InvocationTargetException { + // Test written by Diffblue Cover. + @Test + public void print_json_diffInputNotNullZeroNotNullZeroNotNullOutputIllegalArgumentException3() + throws InvocationTargetException { - // Arrange - final LogBuffer buffer = new LogBuffer(); - buffer.position = 27; - buffer.semival = 0; - final byte[] myByteArray = { - (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, - (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, - (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)1, (byte)0}; - buffer.buffer = myByteArray; - buffer.limit = 31; - buffer.origin = -1; - final long len = 0L; - final String columnName = "foo"; - final int columnIndex = 0; - final String charsetName = "foo"; - try { + // Arrange + final LogBuffer buffer = new LogBuffer(); + buffer.position = 27; + buffer.semival = 0; + final byte[] myByteArray = { (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, + (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, + (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, + (byte) 1, (byte) 0 }; + buffer.buffer = myByteArray; + buffer.limit = 31; + buffer.origin = -1; + final long len = 0L; + final String columnName = "foo"; + final int columnIndex = 0; + final String charsetName = "foo"; + try { - // Act - thrown.expect(IllegalArgumentException.class); - JsonDiffConversion.print_json_diff(buffer, len, columnName, columnIndex, charsetName); - } catch (IllegalArgumentException ex) { + // Act + thrown.expect(IllegalArgumentException.class); + JsonDiffConversion.print_json_diff(buffer, len, columnName, columnIndex, charsetName); + } catch (IllegalArgumentException ex) { - // Assert side effects - Assert.assertNotNull(buffer); - Assert.assertEquals(29, buffer.position); - Assert.assertEquals(0, buffer.semival); - Assert.assertArrayEquals(new byte[] {(byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, - (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, - (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, - (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, - (byte)0, (byte)0, (byte)0, (byte)0, (byte)1, (byte)0}, - buffer.buffer); - Assert.assertEquals(31, buffer.limit); - Assert.assertEquals(-1, buffer.origin); - throw ex; + // Assert side effects + Assert.assertNotNull(buffer); + Assert.assertEquals(29, buffer.position); + Assert.assertEquals(0, buffer.semival); + Assert.assertArrayEquals(new byte[] { (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, + (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, + (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, + (byte) 0, (byte) 1, (byte) 0 }, buffer.buffer); + Assert.assertEquals(31, buffer.limit); + Assert.assertEquals(-1, buffer.origin); + throw ex; + } } - } - // Test written by Diffblue Cover. - @Test - public void print_json_diffInputNotNullZeroNotNullZeroNotNullOutputNotNull() { + // Test written by Diffblue Cover. + @Test + public void print_json_diffInputNotNullZeroNotNullZeroNotNullOutputNotNull() { - // Arrange - final LogBuffer buffer = new LogBuffer(); - final long len = 0L; - final String columnName = ","; - final int columnIndex = 0; - final String charsetName = "1a 2b 3c"; + // Arrange + final LogBuffer buffer = new LogBuffer(); + final long len = 0L; + final String columnName = ","; + final int columnIndex = 0; + final String charsetName = "1a 2b 3c"; - // Act - final StringBuilder actual = - JsonDiffConversion.print_json_diff(buffer, len, columnName, columnIndex, charsetName); + // Act + final StringBuilder actual = JsonDiffConversion.print_json_diff(buffer, + len, + columnName, + columnIndex, + charsetName); - // Assert result - Assert.assertNotNull(actual); - Assert.assertEquals(",", actual.toString()); - } + // Assert result + Assert.assertNotNull(actual); + Assert.assertEquals(",", actual.toString()); + } } diff --git a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/LogEventTest.java b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/LogEventTest.java index fddb063e..bd38816f 100644 --- a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/LogEventTest.java +++ b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/LogEventTest.java @@ -1,6 +1,5 @@ package com.taobao.tddl.dbsync.binlog; -import com.taobao.tddl.dbsync.binlog.LogEvent; import org.junit.Assert; import org.junit.Rule; import org.junit.Test; @@ -9,526 +8,529 @@ import org.junit.rules.Timeout; public class LogEventTest { - @Rule public final ExpectedException thrown = ExpectedException.none(); + @Rule + public final ExpectedException thrown = ExpectedException.none(); - @Rule public final Timeout globalTimeout = new Timeout(10000); + @SuppressWarnings("deprecation") + @Rule + public final Timeout globalTimeout = new Timeout(10000); - /* testedClasses: LogEvent */ - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull() { + /* testedClasses: LogEvent */ + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull() { - // Arrange - final int type = 8; + // Arrange + final int type = 8; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Create_file", actual); - } + // Assert result + Assert.assertEquals("Create_file", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull2() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull2() { - // Arrange - final int type = 12; + // Arrange + final int type = 12; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("New_load", actual); - } + // Assert result + Assert.assertEquals("New_load", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull3() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull3() { - // Arrange - final int type = 39; + // Arrange + final int type = 39; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Update_rows_partial", actual); - } + // Assert result + Assert.assertEquals("Update_rows_partial", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull4() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull4() { - // Arrange - final int type = 36; + // Arrange + final int type = 36; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Unknown", actual); - } + // Assert result + Assert.assertEquals("Unknown", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull5() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull5() { - // Arrange - final int type = 7; + // Arrange + final int type = 7; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Slave", actual); - } + // Assert result + Assert.assertEquals("Slave", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull6() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull6() { - // Arrange - final int type = 6; + // Arrange + final int type = 6; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Load", actual); - } + // Assert result + Assert.assertEquals("Load", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull7() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull7() { - // Arrange - final int type = 10; + // Arrange + final int type = 10; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Exec_load", actual); - } + // Assert result + Assert.assertEquals("Exec_load", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull8() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull8() { - // Arrange - final int type = 5; + // Arrange + final int type = 5; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Intvar", actual); - } + // Assert result + Assert.assertEquals("Intvar", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull9() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull9() { - // Arrange - final int type = 11; + // Arrange + final int type = 11; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Delete_file", actual); - } + // Assert result + Assert.assertEquals("Delete_file", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull10() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull10() { - // Arrange - final int type = 4; + // Arrange + final int type = 4; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Rotate", actual); - } + // Assert result + Assert.assertEquals("Rotate", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull11() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull11() { - // Arrange - final int type = 13; + // Arrange + final int type = 13; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("RAND", actual); - } + // Assert result + Assert.assertEquals("RAND", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull12() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull12() { - // Arrange - final int type = 34; + // Arrange + final int type = 34; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Anonymous_Gtid", actual); - } + // Assert result + Assert.assertEquals("Anonymous_Gtid", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull13() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull13() { - // Arrange - final int type = 15; + // Arrange + final int type = 15; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Format_desc", actual); - } + // Assert result + Assert.assertEquals("Format_desc", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull14() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull14() { - // Arrange - final int type = 32; + // Arrange + final int type = 32; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Delete_rows", actual); - } + // Assert result + Assert.assertEquals("Delete_rows", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull15() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull15() { - // Arrange - final int type = 16; + // Arrange + final int type = 16; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Xid", actual); - } + // Assert result + Assert.assertEquals("Xid", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull16() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull16() { - // Arrange - final int type = 30; + // Arrange + final int type = 30; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Write_rows", actual); - } + // Assert result + Assert.assertEquals("Write_rows", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull17() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull17() { - // Arrange - final int type = 18; + // Arrange + final int type = 18; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Execute_load_query", actual); - } + // Assert result + Assert.assertEquals("Execute_load_query", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull18() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull18() { - // Arrange - final int type = 29; + // Arrange + final int type = 29; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Rows_query", actual); - } + // Assert result + Assert.assertEquals("Rows_query", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull19() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull19() { - // Arrange - final int type = 19; + // Arrange + final int type = 19; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Table_map", actual); - } + // Assert result + Assert.assertEquals("Table_map", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull20() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull20() { - // Arrange - final int type = 28; + // Arrange + final int type = 28; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Ignorable", actual); - } + // Assert result + Assert.assertEquals("Ignorable", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull21() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull21() { - // Arrange - final int type = 33; + // Arrange + final int type = 33; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Gtid", actual); - } + // Assert result + Assert.assertEquals("Gtid", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull22() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull22() { - // Arrange - final int type = 27; + // Arrange + final int type = 27; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Heartbeat", actual); - } + // Assert result + Assert.assertEquals("Heartbeat", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull23() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull23() { - // Arrange - final int type = 26; + // Arrange + final int type = 26; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Incident", actual); - } + // Assert result + Assert.assertEquals("Incident", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull24() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull24() { - // Arrange - final int type = 31; + // Arrange + final int type = 31; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Update_rows", actual); - } + // Assert result + Assert.assertEquals("Update_rows", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull25() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull25() { - // Arrange - final int type = 25; + // Arrange + final int type = 25; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Delete_rows_v1", actual); - } + // Assert result + Assert.assertEquals("Delete_rows_v1", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull26() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull26() { - // Arrange - final int type = 24; + // Arrange + final int type = 24; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Update_rows_v1", actual); - } + // Assert result + Assert.assertEquals("Update_rows_v1", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull27() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull27() { - // Arrange - final int type = 23; + // Arrange + final int type = 23; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Write_rows_v1", actual); - } + // Assert result + Assert.assertEquals("Write_rows_v1", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull28() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull28() { - // Arrange - final int type = 35; + // Arrange + final int type = 35; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Previous_gtids", actual); - } + // Assert result + Assert.assertEquals("Previous_gtids", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull29() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull29() { - // Arrange - final int type = 22; + // Arrange + final int type = 22; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Delete_rows_event_old", actual); - } + // Assert result + Assert.assertEquals("Delete_rows_event_old", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull30() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull30() { - // Arrange - final int type = 21; + // Arrange + final int type = 21; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Update_rows_event_old", actual); - } + // Assert result + Assert.assertEquals("Update_rows_event_old", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull31() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull31() { - // Arrange - final int type = 20; + // Arrange + final int type = 20; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Write_rows_event_old", actual); - } + // Assert result + Assert.assertEquals("Write_rows_event_old", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull32() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull32() { - // Arrange - final int type = 17; + // Arrange + final int type = 17; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Begin_load_query", actual); - } + // Assert result + Assert.assertEquals("Begin_load_query", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull33() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull33() { - // Arrange - final int type = 14; + // Arrange + final int type = 14; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("User var", actual); - } + // Assert result + Assert.assertEquals("User var", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull34() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull34() { - // Arrange - final int type = 9; + // Arrange + final int type = 9; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Append_block", actual); - } + // Assert result + Assert.assertEquals("Append_block", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull35() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull35() { - // Arrange - final int type = 2; + // Arrange + final int type = 2; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Query", actual); - } + // Assert result + Assert.assertEquals("Query", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull36() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull36() { - // Arrange - final int type = 3; + // Arrange + final int type = 3; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Stop", actual); - } + // Assert result + Assert.assertEquals("Stop", actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeNameInputPositiveOutputNotNull37() { + // Test written by Diffblue Cover. + @Test + public void getTypeNameInputPositiveOutputNotNull37() { - // Arrange - final int type = 1; + // Arrange + final int type = 1; - // Act - final String actual = LogEvent.getTypeName(type); + // Act + final String actual = LogEvent.getTypeName(type); - // Assert result - Assert.assertEquals("Start_v3", actual); - } + // Assert result + Assert.assertEquals("Start_v3", actual); + } } diff --git a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/event/LogHeaderTest.java b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/event/LogHeaderTest.java index f9a8d1f5..8392b07f 100644 --- a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/event/LogHeaderTest.java +++ b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/event/LogHeaderTest.java @@ -1,231 +1,233 @@ package com.taobao.tddl.dbsync.binlog.event; -import com.taobao.tddl.dbsync.binlog.event.LogHeader; +import java.util.HashMap; + import org.junit.Assert; import org.junit.Rule; import org.junit.Test; import org.junit.rules.ExpectedException; import org.junit.rules.Timeout; -import java.util.HashMap; - public class LogHeaderTest { - @Rule public final ExpectedException thrown = ExpectedException.none(); + @Rule + public final ExpectedException thrown = ExpectedException.none(); - @Rule public final Timeout globalTimeout = new Timeout(10000); + @SuppressWarnings("deprecation") + @Rule + public final Timeout globalTimeout = new Timeout(10000); - /* testedClasses: LogHeader */ - // Test written by Diffblue Cover. + /* testedClasses: LogHeader */ + // Test written by Diffblue Cover. - @Test - public void constructorInputZeroOutputVoid() { + @Test + public void constructorInputZeroOutputVoid() { - // Arrange - final int type = 0; + // Arrange + final int type = 0; - // Act, creating object to test constructor - final LogHeader objectUnderTest = new LogHeader(type); + // Act, creating object to test constructor + final LogHeader objectUnderTest = new LogHeader(type); - // Assert side effects - final HashMap hashMap = new HashMap(); - Assert.assertEquals(hashMap, objectUnderTest.gtidMap); - } + // Assert side effects + final HashMap hashMap = new HashMap(); + Assert.assertEquals(hashMap, objectUnderTest.gtidMap); + } - // Test written by Diffblue Cover. - @Test - public void getChecksumAlgOutputZero() { + // Test written by Diffblue Cover. + @Test + public void getChecksumAlgOutputZero() { - // Arrange - final LogHeader objectUnderTest = new LogHeader(0); + // Arrange + final LogHeader objectUnderTest = new LogHeader(0); - // Act - final int actual = objectUnderTest.getChecksumAlg(); + // Act + final int actual = objectUnderTest.getChecksumAlg(); - // Assert result - Assert.assertEquals(0, actual); - } + // Assert result + Assert.assertEquals(0, actual); + } - // Test written by Diffblue Cover. - @Test - public void getCrcOutputZero() { + // Test written by Diffblue Cover. + @Test + public void getCrcOutputZero() { - // Arrange - final LogHeader objectUnderTest = new LogHeader(0); + // Arrange + final LogHeader objectUnderTest = new LogHeader(0); - // Act - final long actual = objectUnderTest.getCrc(); + // Act + final long actual = objectUnderTest.getCrc(); - // Assert result - Assert.assertEquals(0L, actual); - } + // Assert result + Assert.assertEquals(0L, actual); + } - // Test written by Diffblue Cover. - @Test - public void getCurrentGtidLastCommitOutputNull() { + // Test written by Diffblue Cover. + @Test + public void getCurrentGtidLastCommitOutputNull() { - // Arrange - final LogHeader objectUnderTest = new LogHeader(0); + // Arrange + final LogHeader objectUnderTest = new LogHeader(0); - // Act - final String actual = objectUnderTest.getCurrentGtidLastCommit(); + // Act + final String actual = objectUnderTest.getCurrentGtidLastCommit(); - // Assert result - Assert.assertNull(actual); - } + // Assert result + Assert.assertNull(actual); + } - // Test written by Diffblue Cover. - @Test - public void getCurrentGtidOutputNull() { + // Test written by Diffblue Cover. + @Test + public void getCurrentGtidOutputNull() { - // Arrange - final LogHeader objectUnderTest = new LogHeader(0); + // Arrange + final LogHeader objectUnderTest = new LogHeader(0); - // Act - final String actual = objectUnderTest.getCurrentGtid(); + // Act + final String actual = objectUnderTest.getCurrentGtid(); - // Assert result - Assert.assertNull(actual); - } + // Assert result + Assert.assertNull(actual); + } - // Test written by Diffblue Cover. - @Test - public void getCurrentGtidSnOutputNull() { + // Test written by Diffblue Cover. + @Test + public void getCurrentGtidSnOutputNull() { - // Arrange - final LogHeader objectUnderTest = new LogHeader(0); + // Arrange + final LogHeader objectUnderTest = new LogHeader(0); - // Act - final String actual = objectUnderTest.getCurrentGtidSn(); + // Act + final String actual = objectUnderTest.getCurrentGtidSn(); - // Assert result - Assert.assertNull(actual); - } + // Assert result + Assert.assertNull(actual); + } - // Test written by Diffblue Cover. - @Test - public void getEventLenOutputZero() { + // Test written by Diffblue Cover. + @Test + public void getEventLenOutputZero() { - // Arrange - final LogHeader objectUnderTest = new LogHeader(0); + // Arrange + final LogHeader objectUnderTest = new LogHeader(0); - // Act - final int actual = objectUnderTest.getEventLen(); + // Act + final int actual = objectUnderTest.getEventLen(); - // Assert result - Assert.assertEquals(0, actual); - } + // Assert result + Assert.assertEquals(0, actual); + } - // Test written by Diffblue Cover. - @Test - public void getFlagsOutputZero() { + // Test written by Diffblue Cover. + @Test + public void getFlagsOutputZero() { - // Arrange - final LogHeader objectUnderTest = new LogHeader(0); + // Arrange + final LogHeader objectUnderTest = new LogHeader(0); - // Act - final int actual = objectUnderTest.getFlags(); + // Act + final int actual = objectUnderTest.getFlags(); - // Assert result - Assert.assertEquals(0, actual); - } + // Assert result + Assert.assertEquals(0, actual); + } - // Test written by Diffblue Cover. - @Test - public void getGtidSetStrOutputNull() { + // Test written by Diffblue Cover. + @Test + public void getGtidSetStrOutputNull() { - // Arrange - final LogHeader objectUnderTest = new LogHeader(0); + // Arrange + final LogHeader objectUnderTest = new LogHeader(0); - // Act - final String actual = objectUnderTest.getGtidSetStr(); + // Act + final String actual = objectUnderTest.getGtidSetStr(); - // Assert result - Assert.assertNull(actual); - } + // Assert result + Assert.assertNull(actual); + } - // Test written by Diffblue Cover. - @Test - public void getLogFileNameOutputNull() { + // Test written by Diffblue Cover. + @Test + public void getLogFileNameOutputNull() { - // Arrange - final LogHeader objectUnderTest = new LogHeader(0); + // Arrange + final LogHeader objectUnderTest = new LogHeader(0); - // Act - final String actual = objectUnderTest.getLogFileName(); + // Act + final String actual = objectUnderTest.getLogFileName(); - // Assert result - Assert.assertNull(actual); - } + // Assert result + Assert.assertNull(actual); + } - // Test written by Diffblue Cover. - @Test - public void getLogPosOutputZero() { + // Test written by Diffblue Cover. + @Test + public void getLogPosOutputZero() { - // Arrange - final LogHeader objectUnderTest = new LogHeader(0); + // Arrange + final LogHeader objectUnderTest = new LogHeader(0); - // Act - final long actual = objectUnderTest.getLogPos(); + // Act + final long actual = objectUnderTest.getLogPos(); - // Assert result - Assert.assertEquals(0L, actual); - } + // Assert result + Assert.assertEquals(0L, actual); + } - // Test written by Diffblue Cover. - @Test - public void getServerIdOutputZero() { + // Test written by Diffblue Cover. + @Test + public void getServerIdOutputZero() { - // Arrange - final LogHeader objectUnderTest = new LogHeader(0); + // Arrange + final LogHeader objectUnderTest = new LogHeader(0); - // Act - final long actual = objectUnderTest.getServerId(); + // Act + final long actual = objectUnderTest.getServerId(); - // Assert result - Assert.assertEquals(0L, actual); - } + // Assert result + Assert.assertEquals(0L, actual); + } - // Test written by Diffblue Cover. - @Test - public void getTypeOutputZero() { + // Test written by Diffblue Cover. + @Test + public void getTypeOutputZero() { - // Arrange - final LogHeader objectUnderTest = new LogHeader(0); + // Arrange + final LogHeader objectUnderTest = new LogHeader(0); - // Act - final int actual = objectUnderTest.getType(); + // Act + final int actual = objectUnderTest.getType(); - // Assert result - Assert.assertEquals(0, actual); - } + // Assert result + Assert.assertEquals(0, actual); + } - // Test written by Diffblue Cover. - @Test - public void getWhenOutputZero() { + // Test written by Diffblue Cover. + @Test + public void getWhenOutputZero() { - // Arrange - final LogHeader objectUnderTest = new LogHeader(0); + // Arrange + final LogHeader objectUnderTest = new LogHeader(0); - // Act - final long actual = objectUnderTest.getWhen(); + // Act + final long actual = objectUnderTest.getWhen(); - // Assert result - Assert.assertEquals(0L, actual); - } + // Assert result + Assert.assertEquals(0L, actual); + } - // Test written by Diffblue Cover. - @Test - public void setLogFileNameInputNotNullOutputVoid() { + // Test written by Diffblue Cover. + @Test + public void setLogFileNameInputNotNullOutputVoid() { - // Arrange - final LogHeader objectUnderTest = new LogHeader(0); - final String logFileName = "3"; + // Arrange + final LogHeader objectUnderTest = new LogHeader(0); + final String logFileName = "3"; - // Act - objectUnderTest.setLogFileName(logFileName); + // Act + objectUnderTest.setLogFileName(logFileName); - // Assert side effects - Assert.assertEquals("3", objectUnderTest.getLogFileName()); - } + // Assert side effects + Assert.assertEquals("3", objectUnderTest.getLogFileName()); + } } diff --git a/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/RegisterSlaveCommandPacketTest.java b/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/RegisterSlaveCommandPacketTest.java index b556a98d..48bb5d3b 100644 --- a/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/RegisterSlaveCommandPacketTest.java +++ b/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/RegisterSlaveCommandPacketTest.java @@ -1,59 +1,58 @@ package com.alibaba.otter.canal.parse.driver.mysql.packets.client; -import com.alibaba.otter.canal.parse.driver.mysql.packets.client.RegisterSlaveCommandPacket; +import java.io.IOException; +import java.lang.reflect.InvocationTargetException; + import org.junit.Assert; import org.junit.Rule; import org.junit.Test; import org.junit.rules.ExpectedException; import org.junit.rules.Timeout; -import java.io.IOException; -import java.lang.reflect.Array; -import java.lang.reflect.InvocationTargetException; - public class RegisterSlaveCommandPacketTest { - @Rule public final ExpectedException thrown = ExpectedException.none(); + @Rule + public final ExpectedException thrown = ExpectedException.none(); - @Rule public final Timeout globalTimeout = new Timeout(10000); + @SuppressWarnings("deprecation") + @Rule + public final Timeout globalTimeout = new Timeout(10000); - /* testedClasses: RegisterSlaveCommandPacket */ - // Test written by Diffblue Cover. - @Test - public void toBytesOutput27() throws IOException, InvocationTargetException { + /* testedClasses: RegisterSlaveCommandPacket */ + // Test written by Diffblue Cover. + @Test + public void toBytesOutput27() throws IOException, InvocationTargetException { - // Arrange - final RegisterSlaveCommandPacket objectUnderTest = new RegisterSlaveCommandPacket(); - objectUnderTest.serverId = 0L; - objectUnderTest.reportPort = 0; - objectUnderTest.reportPasswd = "foo"; - objectUnderTest.reportHost = "foo"; - objectUnderTest.reportUser = "foo"; - objectUnderTest.setCommand((byte)0); + // Arrange + final RegisterSlaveCommandPacket objectUnderTest = new RegisterSlaveCommandPacket(); + objectUnderTest.serverId = 0L; + objectUnderTest.reportPort = 0; + objectUnderTest.reportPasswd = "foo"; + objectUnderTest.reportHost = "foo"; + objectUnderTest.reportUser = "foo"; + objectUnderTest.setCommand((byte) 0); - // Act - final byte[] actual = objectUnderTest.toBytes(); + // Act + final byte[] actual = objectUnderTest.toBytes(); - // Assert result - Assert.assertArrayEquals( - new byte[] {(byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)3, (byte)102, - (byte)111, (byte)111, (byte)3, (byte)102, (byte)111, (byte)111, (byte)3, - (byte)102, (byte)111, (byte)111, (byte)0, (byte)0, (byte)0, (byte)0, - (byte)0, (byte)0, (byte)0, (byte)0, (byte)0, (byte)0}, - actual); - } + // Assert result + Assert.assertArrayEquals(new byte[] { (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 3, (byte) 102, + (byte) 111, (byte) 111, (byte) 3, (byte) 102, (byte) 111, (byte) 111, (byte) 3, (byte) 102, (byte) 111, + (byte) 111, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, (byte) 0, + (byte) 0 }, actual); + } - // Test written by Diffblue Cover. - @Test - public void toLHInputZeroOutput4() { + // Test written by Diffblue Cover. + @Test + public void toLHInputZeroOutput4() { - // Arrange - final int n = 0; + // Arrange + final int n = 0; - // Act - final byte[] actual = RegisterSlaveCommandPacket.toLH(n); + // Act + final byte[] actual = RegisterSlaveCommandPacket.toLH(n); - // Assert result - Assert.assertArrayEquals(new byte[] {(byte)0, (byte)0, (byte)0, (byte)0}, actual); - } + // Assert result + Assert.assertArrayEquals(new byte[] { (byte) 0, (byte) 0, (byte) 0, (byte) 0 }, actual); + } } diff --git a/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientFlatMessageExample.java b/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientFlatMessageExample.java index 4d58b55b..5abcadfe 100644 --- a/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientFlatMessageExample.java +++ b/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientFlatMessageExample.java @@ -8,7 +8,6 @@ import org.slf4j.LoggerFactory; import org.springframework.util.Assert; import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector; -import com.alibaba.otter.canal.example.kafka.AbstractKafkaTest; import com.alibaba.otter.canal.protocol.FlatMessage; /** diff --git a/filter/src/main/java/com/alibaba/otter/canal/filter/PatternUtils.java b/filter/src/main/java/com/alibaba/otter/canal/filter/PatternUtils.java index 52b192d0..b8650c9a 100644 --- a/filter/src/main/java/com/alibaba/otter/canal/filter/PatternUtils.java +++ b/filter/src/main/java/com/alibaba/otter/canal/filter/PatternUtils.java @@ -8,13 +8,12 @@ import org.apache.oro.text.regex.PatternCompiler; import org.apache.oro.text.regex.Perl5Compiler; import com.google.common.base.Function; -import com.google.common.collect.MapMaker; +import com.google.common.cache.CacheBuilder; import com.google.common.collect.MigrateMap; public class PatternUtils { - @SuppressWarnings("deprecation") - private static Map patterns = MigrateMap.makeComputingMap(new MapMaker().softValues(), + private static Map patterns = MigrateMap.makeComputingMap(CacheBuilder.newBuilder().softValues(), new Function() { public Pattern apply(String pattern) { diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java index 49030fc1..9752645a 100644 --- a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java @@ -109,6 +109,7 @@ public class CanalInstanceWithManager extends AbstractCanalInstance { super.start(); } + @SuppressWarnings("resource") protected void initAlarmHandler() { logger.info("init alarmHandler begin..."); String alarmHandlerClass = parameters.getAlarmHandlerClass(); diff --git a/instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/CanalInstanceWithSpring.java b/instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/CanalInstanceWithSpring.java index 69672a6c..8bfaa1e8 100644 --- a/instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/CanalInstanceWithSpring.java +++ b/instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/CanalInstanceWithSpring.java @@ -2,13 +2,12 @@ package com.alibaba.otter.canal.instance.spring; import java.util.List; -import com.alibaba.otter.canal.instance.core.CanalInstance; -import com.alibaba.otter.canal.instance.core.CanalMQConfig; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler; import com.alibaba.otter.canal.instance.core.AbstractCanalInstance; +import com.alibaba.otter.canal.instance.core.CanalMQConfig; import com.alibaba.otter.canal.meta.CanalMetaManager; import com.alibaba.otter.canal.parse.CanalEventParser; import com.alibaba.otter.canal.protocol.CanalEntry; @@ -58,7 +57,7 @@ public class CanalInstanceWithSpring extends AbstractCanalInstance { this.alarmHandler = alarmHandler; } - public void setMqConfig(CanalMQConfig mqConfig){ + public void setMqConfig(CanalMQConfig mqConfig) { this.mqConfig = mqConfig; } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java index c106dd47..84b87044 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java @@ -665,6 +665,14 @@ public class DatabaseTableMeta implements TableMetaTSDB { this.fieldBlackFilterMap = fieldBlackFilterMap; } + public Map> getFieldFilterMap() { + return fieldFilterMap; + } + + public Map> getFieldBlackFilterMap() { + return fieldBlackFilterMap; + } + public int getSnapshotInterval() { return snapshotInterval; } @@ -693,8 +701,4 @@ public class DatabaseTableMeta implements TableMetaTSDB { return false; } - public static void main(String[] args) { - String str = StringUtils.substringBefore("int(11)", "("); - System.out.println(str); - } } diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/SinkCollector.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/SinkCollector.java index 61dc4d64..98f238fc 100644 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/SinkCollector.java +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/SinkCollector.java @@ -1,14 +1,8 @@ package com.alibaba.otter.canal.prometheus.impl; -import com.alibaba.otter.canal.instance.core.CanalInstance; -import com.alibaba.otter.canal.prometheus.InstanceRegistry; -import com.alibaba.otter.canal.sink.CanalEventSink; -import com.alibaba.otter.canal.sink.entry.EntryEventSink; -import com.google.common.base.Preconditions; +import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABELS_LIST; import io.prometheus.client.Collector; import io.prometheus.client.CounterMetricFamily; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.util.ArrayList; import java.util.Collections; @@ -17,8 +11,14 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicLong; -import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABELS_LIST; -import static java.util.concurrent.TimeUnit.NANOSECONDS; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.alibaba.otter.canal.instance.core.CanalInstance; +import com.alibaba.otter.canal.prometheus.InstanceRegistry; +import com.alibaba.otter.canal.sink.CanalEventSink; +import com.alibaba.otter.canal.sink.entry.EntryEventSink; +import com.google.common.base.Preconditions; /** * @author Chuanyi Li @@ -31,9 +31,11 @@ public class SinkCollector extends Collector implements InstanceRegistry { private static final String SINK_BLOCK_TIME_HELP = "Total sink blocking time in milliseconds"; private final ConcurrentMap instances = new ConcurrentHashMap(); - private SinkCollector() {} + private SinkCollector(){ + } private static class SingletonHolder { + private static final SinkCollector SINGLETON = new SinkCollector(); } @@ -45,7 +47,8 @@ public class SinkCollector extends Collector implements InstanceRegistry { public List collect() { List mfs = new ArrayList(); CounterMetricFamily blockingCounter = new CounterMetricFamily(SINK_BLOCKING_TIME, - SINK_BLOCK_TIME_HELP, DEST_LABELS_LIST); + SINK_BLOCK_TIME_HELP, + DEST_LABELS_LIST); for (SinkMetricsHolder smh : instances.values()) { blockingCounter.addMetric(smh.destLabelValues, (smh.eventsSinkBlockingTime.doubleValue() / NANO_PER_MILLI)); } @@ -78,6 +81,7 @@ public class SinkCollector extends Collector implements InstanceRegistry { } private class SinkMetricsHolder { + private AtomicLong eventsSinkBlockingTime; private List destLabelValues; }