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

format

Browse Source
This commit is contained in:
agapple
2019-10-15 21:38:48 +08:00
parent dd7554374f
commit 1fa9020599
3 changed files with 4 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
deployer/src/main/resources/canal.properties
View File

@@ -136,7 +136,6 @@ canal.mq.exchange=
canal.mq.username= canal.mq.username=
canal.mq.password= canal.mq.password=
canal.mq.aliyunuid= canal.mq.aliyunuid=
canal.mq.database.hash = true canal.mq.database.hash = true
################################################## ##################################################
######### Kafka Kerberos Info ############# ######### Kafka Kerberos Info #############

7
server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
View File

@@ -42,14 +42,11 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro
private Producer<String, byte[]> producer; private Producer<String, byte[]> producer;
private MQProperties kafkaProperties; private MQProperties kafkaProperties;
private boolean databaseHash;
@Override @Override
public void init(MQProperties kafkaProperties) { public void init(MQProperties kafkaProperties) {
super.init(kafkaProperties); super.init(kafkaProperties);
this.kafkaProperties = kafkaProperties; this.kafkaProperties = kafkaProperties;
databaseHash = kafkaProperties.getDatabaseHash();
Properties properties = new Properties(); Properties properties = new Properties();
properties.put("bootstrap.servers", kafkaProperties.getServers()); properties.put("bootstrap.servers", kafkaProperties.getServers());
properties.put("acks", kafkaProperties.getAcks()); properties.put("acks", kafkaProperties.getAcks());
@@ -176,7 +173,7 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro
message.getId(), message.getId(),
canalDestination.getPartitionsNum(), canalDestination.getPartitionsNum(),
canalDestination.getPartitionHash(), canalDestination.getPartitionHash(),
databaseHash); kafkaProperties.getDatabaseHash());
int length = messages.length; int length = messages.length;
for (int i = 0; i < length; i++) { for (int i = 0; i < length; i++) {
Message messagePartition = messages[i]; Message messagePartition = messages[i];
@@ -206,7 +203,7 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro
FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage, FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage,
canalDestination.getPartitionsNum(), canalDestination.getPartitionsNum(),
canalDestination.getPartitionHash(), canalDestination.getPartitionHash(),
databaseHash); kafkaProperties.getDatabaseHash());
int length = partitionFlatMessage.length; int length = partitionFlatMessage.length;
for (int i = 0; i < length; i++) { for (int i = 0; i < length; i++) {
FlatMessage flatMessagePart = partitionFlatMessage[i]; FlatMessage flatMessagePart = partitionFlatMessage[i];

7
server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java
View File

@@ -40,13 +40,10 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
private MQProperties mqProperties; private MQProperties mqProperties;
private static final String CLOUD_ACCESS_CHANNEL = "cloud"; private static final String CLOUD_ACCESS_CHANNEL = "cloud";
private boolean databaseHash;
@Override @Override
public void init(MQProperties rocketMQProperties) { public void init(MQProperties rocketMQProperties) {
super.init(rocketMQProperties); super.init(rocketMQProperties);
this.mqProperties = rocketMQProperties; this.mqProperties = rocketMQProperties;
this.databaseHash = rocketMQProperties.getDatabaseHash();
RPCHook rpcHook = null; RPCHook rpcHook = null;
if (rocketMQProperties.getAliyunAccessKey().length() > 0 if (rocketMQProperties.getAliyunAccessKey().length() > 0
&& rocketMQProperties.getAliyunSecretKey().length() > 0) { && rocketMQProperties.getAliyunSecretKey().length() > 0) {
@@ -129,7 +126,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
message.getId(), message.getId(),
destination.getPartitionsNum(), destination.getPartitionsNum(),
destination.getPartitionHash(), destination.getPartitionHash(),
databaseHash); mqProperties.getDatabaseHash());
int length = messages.length; int length = messages.length;
ExecutorTemplate template = new ExecutorTemplate(executor); ExecutorTemplate template = new ExecutorTemplate(executor);
@@ -173,7 +170,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage, FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage,
destination.getPartitionsNum(), destination.getPartitionsNum(),
destination.getPartitionHash(), destination.getPartitionHash(),
databaseHash); mqProperties.getDatabaseHash());
int length = partitionFlatMessage.length; int length = partitionFlatMessage.length;
for (int i = 0; i < length; i++) { for (int i = 0; i < length; i++) {
partitionFlatMessages.get(i).add(partitionFlatMessage[i]); partitionFlatMessages.get(i).add(partitionFlatMessage[i]);