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(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 record; if (topic.getPartition() != null) { record = new ProducerRecord(topic.getTopic(), topic.getPartition(), null, message); @@ -59,13 +59,13 @@ public class CanalKafkaProducer { record = new ProducerRecord(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(); } }