From 0cda90cdeae57c4d30468ffa581e105c784695da Mon Sep 17 00:00:00 2001
From: mcy <rewerma@163.com>
Date: Thu, 20 Sep 2018 16:17:38 +0800
Subject: [PATCH] =?UTF-8?q?=E9=BB=98=E8=AE=A4=E5=B0=86=E6=95=B0=E6=8D=AE?=
 =?UTF-8?q?=E5=8F=91=E7=8E=B0kafka=E7=9A=84partition=200=EF=BC=8C=E9=81=BF?=
 =?UTF-8?q?=E5=85=8D=E5=A4=9A=E5=88=86=E5=8C=BA=E7=9A=84=E6=95=B0=E6=8D=AE?=
 =?UTF-8?q?=E4=B8=80=E8=87=B4=E6=80=A7=E9=97=AE=E9=A2=98=20=E5=90=8E?=
 =?UTF-8?q?=E7=BB=AD=E5=B0=86=E6=94=AF=E6=8C=81=E9=80=9A=E8=BF=87id=20hash?=
 =?UTF-8?q?=E5=88=86=E5=8F=91=E6=95=B0=E6=8D=AE?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../otter/canal/kafka/CanalKafkaProducer.java    | 16 +++++++++++++---
 1 file changed, 13 insertions(+), 3 deletions(-)

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