chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' of github.com:alibaba/canal

Browse Source
This commit is contained in:
wanshao
2020-02-28 14:24:33 +08:00
parent c21c18458b 5e761e14fa
commit beaca165d0
51 changed files with 1596 additions and 1361 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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 外部适配器

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

@@ -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<String, Object> 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<String, Object> 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) {

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

@@ -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;

19
client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/support/ESConnection.java
View File

@@ -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();

6
client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/LabelSyncJoinSub2Test.java
View File

@@ -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;

6
client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/LabelSyncJoinSubTest.java
View File

@@ -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;

6
client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/RoleSyncJoinOne2Test.java
View File

@@ -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;

6
client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/RoleSyncJoinOneTest.java
View File

@@ -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;

6
client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/UserSyncJoinOneTest.java
View File

@@ -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;

6
client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/sync/UserSyncSingleTest.java
View File

@@ -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;

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

@@ -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 外部适配器
*

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

@@ -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;

10
client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/support/ESConnection.java
View File

@@ -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);
}

6
client-adapter/es7x/src/test/java/com/alibaba/otter/canal/client/adapter/es7x/test/ES7xTest.java
View File

@@ -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);
}

2
client-adapter/es7x/src/test/java/com/alibaba/otter/canal/client/adapter/es7x/test/ESConnectionTest.java
View File

@@ -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 {

57
client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/ESAdapter.java
View File

@@ -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<String, ESSyncConfig> 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<String, ESSyncConfig> 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<String, ESSyncConfig> 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);
});
}
}

15
client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/SqlParser.java
View File

@@ -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;

27
client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/monitor/ESConfigMonitor.java
View File

@@ -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;
}

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

@@ -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;

8
client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/support/ESSyncUtil.java
View File

@@ -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;

280
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java
View File

@@ -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);

4
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfigHolder.java
View File

@@ -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

8
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AdapterProcessor.java
View File

@@ -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;
/**
* 适配处理器

257
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java
View File

@@ -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<String, AdapterProcessor> canalAdapterProcessors = new HashMap<>();
private Map<String, AdapterProcessor> canalAdapterProcessors = new HashMap<>();
private ExtensionLoader<OuterAdapter> loader;
private ExtensionLoader<OuterAdapter> 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<List<OuterAdapter>> canalOuterAdapterGroups = new CopyOnWriteArrayList<>();
//
// for (CanalClientConfig.Group connectorGroup : canalAdapter.getGroups()) {
// List<OuterAdapter> 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<List<OuterAdapter>> canalOuterAdapterGroups = new CopyOnWriteArrayList<>();
// List<OuterAdapter> 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<List<OuterAdapter>> canalOuterAdapterGroups = new CopyOnWriteArrayList<>();
// List<OuterAdapter> 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<List<OuterAdapter>> canalOuterAdapterGroups = new CopyOnWriteArrayList<>();
// List<OuterAdapter> 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<List<OuterAdapter>> canalOuterAdapterGroups = new
// CopyOnWriteArrayList<>();
//
// for (CanalClientConfig.Group connectorGroup :
// canalAdapter.getGroups()) {
// List<OuterAdapter> 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<List<OuterAdapter>> canalOuterAdapterGroups = new
// CopyOnWriteArrayList<>();
// List<OuterAdapter> 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<List<OuterAdapter>> canalOuterAdapterGroups = new
// CopyOnWriteArrayList<>();
// List<OuterAdapter> 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<List<OuterAdapter>> canalOuterAdapterGroups = new
// CopyOnWriteArrayList<>();
// List<OuterAdapter> 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<OuterAdapter> canalOutConnectors) {

4
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/DbRemoteConfigLoader.java
View File

@@ -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;
/**

3
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteAdapterMonitorImpl.java
View File

@@ -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;
/**
* 远程配置监听器实现
*

35
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java
View File

@@ -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<String, String> targetPk = new LinkedHashMap<>(); // 目标表主键字段
private boolean mapAll = false; // 映射所有字段
private String targetDb; // 目标库名
private String targetTable; // 目标表名
private Map<String, String> targetColumns; // 目标表字段映射
private boolean mirrorDb = false; // 是否镜像库
private String database; // 数据库名或schema名
private String table; // 表名
private Map<String, String> targetPk = new LinkedHashMap<>(); // 目标表主键字段
private boolean mapAll = false; // 映射所有字段
private String targetDb; // 目标库名
private String targetTable; // 目标表名
private Map<String, String> 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<String, String> 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;
}

6
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java
View File

@@ -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;

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

@@ -91,9 +91,10 @@ public class RdbMirrorDbSyncService {
if (config == null) {
return false;
}
// 是否区分大小写
boolean caseInsensitive = config.getDbMapping().isCaseInsensitive();
if (config.getConcurrent()) {
List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml);
List<SingleDml> 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<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml);
List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive);
singleDmls.forEach(singleDml -> {
RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml);
rdbSyncService.getDmlsPartition()[hash].add(syncItem);

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

@@ -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<String, MappingConfig> 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<SingleDml> 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<String, MappingConfig> configMap;
if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
configMap = mappingConfig.get(destination + "-" + groupId + "_" + database + "-" + table);
} else {
int hash = 0;
List<SingleDml> 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<SingleDml> 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<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive);
singleDmls.forEach(singleDml -> {
SyncItem syncItem = new SyncItem(config, singleDml);
dmlsPartition[hash].add(syncItem);
});
}
}
return true;
}
return true;
}
} );
});
}
/**

26
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SingleDml.java
View File

@@ -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<SingleDml> dml2SingleDmls(Dml dml) {
public static List<SingleDml> dml2SingleDmls(Dml dml, boolean caseInsensitive) {
List<SingleDml> 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<String, Object> data = dml.getData().get(i);
if (caseInsensitive) {
data = toCaseInsensitiveMap(data);
}
singleDml.setData(data);
if (dml.getOld() != null) {
singleDml.setOld(dml.getOld().get(i));
Map<String, Object> 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<SingleDml> dml2SingleDmls(Dml dml) {
return dml2SingleDmls(dml, false);
}
private static <V> LinkedCaseInsensitiveMap<V> toCaseInsensitiveMap(Map<String, V> data) {
LinkedCaseInsensitiveMap map = new LinkedCaseInsensitiveMap();
map.putAll(data);
return map;
}
}

11
client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/DBTest.java
View File

@@ -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)) {

1
client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java
View File

@@ -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;

125
common/src/main/java/com/google/common/collect/MigrateMap.java
View File

@@ -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 <K, V> ConcurrentMap<K, V> makeComputingMap(MapMaker maker,
public static <K, V> ConcurrentMap<K, V> makeComputingMap(CacheBuilder<Object, Object> builder,
Function<? super K, ? extends V> computingFunction) {
return maker.makeComputingMap(computingFunction);
final Function<? super K, ? extends V> function = computingFunction;
LoadingCache<K, V> computingCache = builder.build(new CacheLoader<K, V>() {
@Override
public V load(K key) throws Exception {
return function.apply(key);
}
});
return new MigrateConcurrentMap<K, V>(computingCache);
}
@SuppressWarnings("deprecation")
public static <K, V> ConcurrentMap<K, V> makeComputingMap(Function<? super K, ? extends V> computingFunction) {
return new MapMaker().makeComputingMap(computingFunction);
return makeComputingMap(CacheBuilder.newBuilder(), computingFunction);
}
final static class MigrateConcurrentMap<K, V> implements ConcurrentMap<K, V> {
private final LoadingCache<K, V> computingCache;
private final ConcurrentMap<K, V> cacheView;
MigrateConcurrentMap(LoadingCache<K, V> 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<? extends K, ? extends V> m) {
cacheView.putAll(m);
}
@Override
public void clear() {
cacheView.clear();
}
@Override
public Set<K> keySet() {
return cacheView.keySet();
}
@Override
public Collection<V> values() {
return cacheView.values();
}
@Override
public Set<Entry<K, V>> 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();
}
}
}

27
connector/core/src/main/java/com/alibaba/otter/canal/connector/core/consumer/CommonMessage.java
View File

@@ -12,21 +12,18 @@ public class CommonMessage implements Serializable {
private String table; // 表名
private List<String> 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<Map<String, Object>> data; // 数据列表
private List<Map<String, Object>> 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<Map<String, Object>> data;
// 旧数据列表,用于update,size和data的size一一对应
private List<Map<String, Object>> old;
public String getDatabase() {
return database;

5
connector/core/src/main/java/com/alibaba/otter/canal/connector/core/filter/PatternUtils.java
View File

@@ -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<String, Pattern> patterns = MigrateMap.makeComputingMap(new MapMaker().softValues(),
private static Map<String, Pattern> patterns = MigrateMap.makeComputingMap(CacheBuilder.newBuilder().softValues(),
pattern -> {
try {
PatternCompiler pc = new Perl5Compiler();

10
connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQMessageUtils.java
View File

@@ -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<String, List<PartitionData>> partitionDatas = MigrateMap.makeComputingMap(new MapMaker().softValues(),
private static Map<String, List<PartitionData>> partitionDatas = MigrateMap.makeComputingMap(CacheBuilder.newBuilder()
.softValues(),
pkHashConfigs -> {
List<PartitionData> datas = Lists.newArrayList();
@@ -73,8 +73,8 @@ public class MQMessageUtils {
return datas;
});
@SuppressWarnings("deprecation")
private static Map<String, List<DynamicTopicData>> dynamicTopicDatas = MigrateMap.makeComputingMap(new MapMaker().softValues(),
private static Map<String, List<DynamicTopicData>> dynamicTopicDatas = MigrateMap.makeComputingMap(CacheBuilder.newBuilder()
.softValues(),
new Function<String, List<DynamicTopicData>>() {
public List<DynamicTopicData> apply(String pkHashConfigs) {

1
connector/core/src/main/java/com/alibaba/otter/canal/connector/core/spi/ExtensionLoader.java
View File

@@ -409,7 +409,6 @@ public class ExtensionLoader<T> {
}
}
@SuppressWarnings("unused")
private static ClassLoader findClassLoader() {
return ExtensionLoader.class.getClassLoader();
}

10
connector/tcp-connector/src/main/java/com/alibaba/otter/canal/connector/tcp/consumer/CanalTCPConsumer.java
View File

@@ -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) {

2
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java
View File

@@ -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;

64
dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/JsonConversion_Json_ValueTest.java
View File

@@ -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
}
}

349
dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/JsonDiffConversionTest.java
View File

@@ -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());
}
}

750
dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/LogEventTest.java
View File

@@ -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);
}
}

318
dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/event/LogHeaderTest.java
View File

@@ -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<String, String> hashMap = new HashMap<String, String>();
Assert.assertEquals(hashMap, objectUnderTest.gtidMap);
}
// Assert side effects
final HashMap<String, String> hashMap = new HashMap<String, String>();
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());
}
}

77
driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/RegisterSlaveCommandPacketTest.java
View File

@@ -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);
}
}

1
example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientFlatMessageExample.java
View File

@@ -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;
/**

5
filter/src/main/java/com/alibaba/otter/canal/filter/PatternUtils.java
View File

@@ -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<String, Pattern> patterns = MigrateMap.makeComputingMap(new MapMaker().softValues(),
private static Map<String, Pattern> patterns = MigrateMap.makeComputingMap(CacheBuilder.newBuilder().softValues(),
new Function<String, Pattern>() {
public Pattern apply(String pattern) {

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

@@ -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();

5
instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/CanalInstanceWithSpring.java
View File

@@ -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;
}

12
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java
View File

@@ -665,6 +665,14 @@ public class DatabaseTableMeta implements TableMetaTSDB {
this.fieldBlackFilterMap = fieldBlackFilterMap;
}
public Map<String, List<String>> getFieldFilterMap() {
return fieldFilterMap;
}
public Map<String, List<String>> 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);
}
}

26
prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/SinkCollector.java
View File

@@ -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<String, SinkMetricsHolder> instances = new ConcurrentHashMap<String, SinkMetricsHolder>();
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<MetricFamilySamples> collect() {
List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>();
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<String> destLabelValues;
}