chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

0.11的环境未测过，先去除kafka的事务

Browse Source
This commit is contained in:
mcy
2018-09-03 18:07:43 +08:00
parent ebf9139762
commit 17f345bfe5
1 changed files with 4 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
View File

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