From 728550cac3dd9a400f2d0077c363a25b4a3a399b Mon Sep 17 00:00:00 2001 From: mcy Date: Tue, 4 Sep 2018 09:56:51 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0kafka=20producer=20send?= =?UTF-8?q?=E7=BB=93=E6=9E=9C=E5=90=8C=E6=AD=A5=E5=88=A4=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../alibaba/otter/canal/kafka/CanalKafkaProducer.java | 11 +++++++---- 1 file changed, 7 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 ec1a5926..0685f76e 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 @@ -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(topic.getTopic(), message); } - producer.send(record); + Future 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(); }