From 4ebf17f57b8c278809f081b9e59d2c89e2a6b4ef Mon Sep 17 00:00:00 2001
From: machey <rewerma@163.com>
Date: Sun, 20 Jan 2019 14:30:20 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BB=A3=E7=A0=81=E8=B0=83=E6=95=B4?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../otter/canal/kafka/CanalKafkaProducer.java | 58 ++++++++-----------
 1 file changed, 25 insertions(+), 33 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 cf5931ef..1177a82e 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
@@ -140,11 +140,12 @@ public class CanalKafkaProducer implements CanalMQProducer {
             if (flatMessages != null) {
                 // 开启事务，需要kafka版本支持
                 if (kafkaProperties.getTransaction()) {
+                    producer.initTransactions();
                     producer.beginTransaction();
                 }
-                for (FlatMessage flatMessage : flatMessages) {
-                    if (StringUtils.isEmpty(canalDestination.getPartitionHash())) {
-                        try {
+                try {
+                    for (FlatMessage flatMessage : flatMessages) {
+                        if (StringUtils.isEmpty(canalDestination.getPartitionHash())) {
                             Integer partition = canalDestination.getPartition();
                             if (partition == null) {
                                 partition = 0;
@@ -158,21 +159,14 @@ public class CanalKafkaProducer implements CanalMQProducer {
                             } else {
                                 producer2.send(record).get();
                             }
-                        } catch (Exception e) {
-                            if (kafkaProperties.getTransaction()) {
-                                producer.abortTransaction();
-                            }
-                            throw e;
-                        }
-                    } else {
-                        FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage,
-                            canalDestination.getPartitionsNum(),
-                            canalDestination.getPartitionHash());
-                        int length = partitionFlatMessage.length;
-                        for (int i = 0; i < length; i++) {
-                            FlatMessage flatMessagePart = partitionFlatMessage[i];
-                            if (flatMessagePart != null) {
-                                try {
+                        } else {
+                            FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage,
+                                canalDestination.getPartitionsNum(),
+                                canalDestination.getPartitionHash());
+                            int length = partitionFlatMessage.length;
+                            for (int i = 0; i < length; i++) {
+                                FlatMessage flatMessagePart = partitionFlatMessage[i];
+                                if (flatMessagePart != null) {
                                     ProducerRecord<String, String> record = new ProducerRecord<String, String>(
                                         topicName,
                                         i,
@@ -183,30 +177,28 @@ public class CanalKafkaProducer implements CanalMQProducer {
                                     } else {
                                         producer2.send(record).get();
                                     }
-                                } catch (Exception e) {
-                                    if (kafkaProperties.getTransaction()) {
-                                        producer.abortTransaction();
-                                    }
-                                    throw e;
                                 }
                             }
                         }
-                    }
 
+                        if (kafkaProperties.getTransaction()) {
+                            producer.commitTransaction();
+                        }
+
+                        if (logger.isDebugEnabled()) {
+                            logger.debug("Send flat message to kafka topic: [{}], packet: {}",
+                                topicName,
+                                JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue));
+                        }
+                    }
+                } catch (Exception e) {
                     if (kafkaProperties.getTransaction()) {
-                        producer.commitTransaction();
-                    }
-
-                    if (logger.isDebugEnabled()) {
-                        logger.debug("Send flat message to kafka topic: [{}], packet: {}",
-                            topicName,
-                            JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue));
+                        producer.abortTransaction();
                     }
+                    throw e;
                 }
             }
         }
-
-        // producer.commitTransaction();
     }
 
 }