From 17f345bfe5bdf7d7eb00ffe6667573d7ec5bab84 Mon Sep 17 00:00:00 2001
From: mcy <rewerma@163.com>
Date: Mon, 3 Sep 2018 18:07:43 +0800
Subject: [PATCH] =?UTF-8?q?0.11=E7=9A=84=E7=8E=AF=E5=A2=83=E6=9C=AA?=
 =?UTF-8?q?=E6=B5=8B=E8=BF=87=EF=BC=8C=E5=85=88=E5=8E=BB=E9=99=A4kafka?=
 =?UTF-8?q?=E7=9A=84=E4=BA=8B=E5=8A=A1?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../com/alibaba/otter/canal/kafka/CanalKafkaProducer.java | 8 ++++----
 1 file changed, 4 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 ae63f415..ec1a5926 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
@@ -35,6 +35,7 @@ public class CanalKafkaProducer {
         properties.put("key.serializer", StringSerializer.class.getName());
         properties.put("value.serializer", MessageSerializer.class.getName());
         producer = new KafkaProducer<String, Message>(properties);
+        // producer.initTransactions();
     }
 
     public void stop() {
@@ -49,9 +50,8 @@ public class CanalKafkaProducer {
     }
 
     public void send(KafkaProperties.Topic topic, Message message, Callback callback) {
-        producer.initTransactions();
         try {
-            producer.beginTransaction();
+            // producer.beginTransaction();
             ProducerRecord<String, Message> record;
             if (topic.getPartition() != null) {
                 record = new ProducerRecord<String, Message>(topic.getTopic(), topic.getPartition(), null, message);
@@ -59,13 +59,13 @@ public class CanalKafkaProducer {
                 record = new ProducerRecord<String, Message>(topic.getTopic(), message);
             }
             producer.send(record);
-            producer.commitTransaction();
+            // producer.commitTransaction();
             callback.commit();
             if (logger.isDebugEnabled()) {
                 logger.debug("send message to kafka topic: {} \n {}", topic.getTopic(), message.toString());
             }
         } catch (KafkaException e) {
-            producer.abortTransaction();
+            // producer.abortTransaction();
             callback.rollback();
         }
     }