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

默认将数据发现kafka的partition 0，避免多分区的数据一致性问题

Browse Source 
后续将支持通过id hash分发数据
This commit is contained in:
mcy
2018-09-20 16:17:38 +08:00
parent 1a11153cd2
commit 0cda90cdea
1 changed files with 13 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -75,7 +75,7 @@ public class CanalKafkaProducer {
if (topic.getPartition() != null) { if (topic.getPartition() != null) {
record = new ProducerRecord<String, Message>(topic.getTopic(), topic.getPartition(), null, message); record = new ProducerRecord<String, Message>(topic.getTopic(), topic.getPartition(), null, message);
} else { } else {
record = new ProducerRecord<String, Message>(topic.getTopic(), message); record = new ProducerRecord<String, Message>(topic.getTopic(), 0, null, message);
} }
producer.send(record); producer.send(record);
@@ -84,8 +84,18 @@ public class CanalKafkaProducer {
List<FlatMessage> flatMessages = FlatMessage.messageConverter(message); List<FlatMessage> flatMessages = FlatMessage.messageConverter(message);
if (flatMessages != null) { if (flatMessages != null) {
for (FlatMessage flatMessage : flatMessages) { for (FlatMessage flatMessage : flatMessages) {
ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic.getTopic(), ProducerRecord<String, String> record;
if (topic.getPartition() != null) {
record = new ProducerRecord<String, String>(topic.getTopic(),
topic.getPartition(),
null,
JSON.toJSONString(flatMessage)); JSON.toJSONString(flatMessage));
} else {
record = new ProducerRecord<String, String>(topic.getTopic(),
0,
null,
JSON.toJSONString(flatMessage));
}
producer2.send(record); producer2.send(record);
} }
} }