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

增加kafka producer send结果同步判断

Browse Source
This commit is contained in:
mcy
2018-09-04 09:56:51 +08:00
parent 81c216f1c1
commit 728550cac3
1 changed files with 7 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,11 +1,12 @@
package com.alibaba.otter.canal.kafka;
import java.util.Properties;
import java.util.concurrent.Future;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -58,13 +59,15 @@ public class CanalKafkaProducer {
} else {
record = new ProducerRecord<String, Message>(topic.getTopic(), message);
}
producer.send(record);
Future<RecordMetadata> future = producer.send(record);
future.get();
// producer.commitTransaction();
callback.commit();
if (logger.isDebugEnabled()) {
logger.debug("send message to kafka topic: {} \n {}", topic.getTopic(), message.toString());
logger.debug("send message to kafka topic: {}", topic.getTopic());
}
} catch (KafkaException e) {
} catch (Exception e) {
logger.error(e.getMessage(), e);
// producer.abortTransaction();
callback.rollback();
}