diff --git a/deployer/src/main/resources/kafka.yml b/deployer/src/main/resources/kafka.yml index b86b3034..76e2bade 100644 --- a/deployer/src/main/resources/kafka.yml +++ b/deployer/src/main/resources/kafka.yml @@ -12,9 +12,11 @@ flatMessage: true canalDestinations: - canalDestination: example - topic: expample - #partitionsNum: 3 - #partitionHash: - # mytest.person: id + topic: example +# #对应topic分区数量 +# partitionsNum: 3 +# partitionHash: +# #库名.表名: 唯一主键 +# mytest.person: id diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index fbeeeb87..5708105c 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -99,16 +99,15 @@ public class CanalKafkaProducer { canalDestination.getPartitionHash()); int length = partitionFlatMessage.length; for (int i = 0; i < length; i++) { - FlatMessage flatMessage1 = partitionFlatMessage[i]; - if (flatMessage1 == null) { - continue; + FlatMessage flatMessagePart = partitionFlatMessage[i]; + if (flatMessagePart != null) { + ProducerRecord record = new ProducerRecord( + canalDestination.getTopic(), + i, + null, + JSON.toJSONString(flatMessagePart)); + producer2.send(record); } - ProducerRecord record = new ProducerRecord( - canalDestination.getTopic(), - i, - null, - JSON.toJSONString(partitionFlatMessage[i])); - producer2.send(record); } } else { ProducerRecord record = new ProducerRecord(