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