From f9989e085674c0f5068f9568448329fa8513d16e Mon Sep 17 00:00:00 2001
From: nieyuan1991 <nieyuan1991@126.com>
Date: Tue, 28 May 2019 00:02:25 +0800
Subject: [PATCH] =?UTF-8?q?fix:=E6=9C=AA=E6=8D=95=E8=8E=B7kafka=E4=BA=8B?=
 =?UTF-8?q?=E5=8A=A1=E5=BC=82=E5=B8=B8=EF=BC=8C=E9=80=A0=E6=88=90canal=20a?=
 =?UTF-8?q?ck=E5=BC=82=E5=B8=B8=20(#1814)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../otter/canal/kafka/CanalKafkaProducer.java        | 12 ++++++++----
 1 file changed, 8 insertions(+), 4 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 89690a29..51fce634 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
@@ -102,10 +102,10 @@ public class CanalKafkaProducer implements CanalMQProducer {
             producerTmp = producer2;
         }
 
-        if (kafkaProperties.getTransaction()) {
-            producerTmp.beginTransaction();
-        }
         try {
+            if (kafkaProperties.getTransaction()) {
+                producerTmp.beginTransaction();
+            }
             if (!StringUtils.isEmpty(canalDestination.getDynamicTopic())) {
                 // 动态topic
                 Map<String, Message> messageMap = MQMessageUtils.messageTopics(message,
@@ -130,7 +130,11 @@ public class CanalKafkaProducer implements CanalMQProducer {
         } catch (Throwable e) {
             logger.error(e.getMessage(), e);
             if (kafkaProperties.getTransaction()) {
-                producerTmp.abortTransaction();
+                try {
+                    producerTmp.abortTransaction();
+                } catch (Exception e1) {
+                    logger.error(e1.getMessage(), e1);
+                }
             }
             callback.rollback();
         }