wyl/canal
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

fixed issue #1636

Browse Source 
非flat模式，当MQ主题为多分区时，数据拆分只取了for循环里面的最后一份数据，
导致消费端只能消费到一个分区的数据
This commit is contained in:
wangsaner
2019-03-27 21:57:04 +08:00
parent 0ec4699149
commit 739963acd0
1 changed files with 13 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
+13 -14
View File
@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.kafka;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
@@ -138,7 +139,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
private void send(MQProperties.CanalDestination canalDestination, String topicName, Message message)
throws Exception {
if (!kafkaProperties.getFlatMessage()) {
ProducerRecord<String, Message> record = null;
List<ProducerRecord<String, Message>> records = new ArrayList<>();
if (canalDestination.getPartitionHash() != null && !canalDestination.getPartitionHash().isEmpty()) {
Message[] messages = MQMessageUtils.messagePartition(message,
canalDestination.getPartitionsNum(),
@@ -147,25 +148,23 @@ public class CanalKafkaProducer implements CanalMQProducer {
for (int i = 0; i < length; i++) {
Message messagePartition = messages[i];
if (messagePartition != null) {
record = new ProducerRecord<>(topicName, i, null, messagePartition);
records.add(new ProducerRecord<>(topicName, i, null, messagePartition));
}
}
} else {
final int partition = canalDestination.getPartition() != null ? canalDestination.getPartition() : 0;
record = new ProducerRecord<>(topicName, partition, null, message);
records.add(new ProducerRecord<>(topicName, partition, null, message));
}
if (record != null) {
if (kafkaProperties.getTransaction()) {
producer.send(record).get();
} else {
producer.send(record).get();
}
if (logger.isDebugEnabled()) {
logger.debug("Send message to kafka topic: [{}], packet: {}", topicName, message.toString());
}
}
if (!records.isEmpty()) {
for (ProducerRecord<String, Message> record : records) {
producer.send(record).get();
}
if (logger.isDebugEnabled()) {
logger.debug("Send message to kafka topic: [{}], packet: {}", topicName, message.toString());
}
}
} else {
// 发送扁平数据json
List<FlatMessage> flatMessages = MQMessageUtils.messageConverter(message);