From 739963acd03c9633416ccca85a80ef1cc8a6ff40 Mon Sep 17 00:00:00 2001
From: wangsaner <snakeliu@reabam.com>
Date: Wed, 27 Mar 2019 21:57:04 +0800
Subject: [PATCH] =?UTF-8?q?fixed=20issue=20#1636=20=E9=9D=9Eflat=E6=A8=A1?=
 =?UTF-8?q?=E5=BC=8F=EF=BC=8C=E5=BD=93MQ=E4=B8=BB=E9=A2=98=E4=B8=BA?=
 =?UTF-8?q?=E5=A4=9A=E5=88=86=E5=8C=BA=E6=97=B6=EF=BC=8C=E6=95=B0=E6=8D=AE?=
 =?UTF-8?q?=E6=8B=86=E5=88=86=E5=8F=AA=E5=8F=96=E4=BA=86for=E5=BE=AA?=
 =?UTF-8?q?=E7=8E=AF=E9=87=8C=E9=9D=A2=E7=9A=84=E6=9C=80=E5=90=8E=E4=B8=80?=
 =?UTF-8?q?=E4=BB=BD=E6=95=B0=E6=8D=AE=EF=BC=8C=20=E5=AF=BC=E8=87=B4?=
 =?UTF-8?q?=E6=B6=88=E8=B4=B9=E7=AB=AF=E5=8F=AA=E8=83=BD=E6=B6=88=E8=B4=B9?=
 =?UTF-8?q?=E5=88=B0=E4=B8=80=E4=B8=AA=E5=88=86=E5=8C=BA=E7=9A=84=E6=95=B0?=
 =?UTF-8?q?=E6=8D=AE?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../otter/canal/kafka/CanalKafkaProducer.java | 27 +++++++++----------
 1 file changed, 13 insertions(+), 14 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 a3a8aec5..c4605067 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
@@ -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);