From ee5775ce05acfc053842a69f24ef745c76c74171 Mon Sep 17 00:00:00 2001 From: machey Date: Sun, 2 Sep 2018 20:21:07 +0800 Subject: [PATCH 1/5] =?UTF-8?q?=E5=8E=BB=E6=8E=89kafka=20server=E7=AB=AFTR?= =?UTF-8?q?ANSACTIONBEGIN=E5=92=8CTRANSACTIONEND=E7=9A=84=E5=88=A4?= =?UTF-8?q?=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../client/kafka/KafkaCanalConnector.java | 2 +- .../otter/canal/kafka/CanalKafkaProducer.java | 28 +------------------ .../otter/canal/kafka/CanalKafkaStarter.java | 6 ++-- 3 files changed, 5 insertions(+), 31 deletions(-) diff --git a/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java index c5b2f8fc..7afb9c09 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java @@ -45,7 +45,7 @@ public class KafkaCanalConnector { properties.put("auto.offset.reset", "latest"); // 如果没有offset则从最后的offset开始读 properties.put("request.timeout.ms", "40000"); // 必须大于session.timeout.ms的设置 properties.put("session.timeout.ms", "30000"); // 默认为30秒 - properties.put("max.poll.records", "1"); // 所以一次只取一条数据 + properties.put("max.poll.records", "1"); // 一次只取一条message数据 properties.put("key.deserializer", StringDeserializer.class.getName()); properties.put("value.deserializer", MessageDeserializer.class.getName()); 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 b6dba051..a6d084e4 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 @@ -48,33 +48,7 @@ public class CanalKafkaProducer { } } - public void send(KafkaProperties.Topic topic, Message message) throws IOException { - // set canal.instance.filter.transaction.entry = true - - // boolean valid = false; - // if (message != null) { - // if (message.isRaw() && !message.getRawEntries().isEmpty()) { - // for (ByteString byteString : message.getRawEntries()) { - // CanalEntry.Entry entry = CanalEntry.Entry.parseFrom(byteString); - // if (entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONBEGIN - // && entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONEND) { - // valid = true; - // break; - // } - // } - // } else if (!message.getEntries().isEmpty()){ - // for (CanalEntry.Entry entry : message.getEntries()) { - // if (entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONBEGIN - // && entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONEND) { - // valid = true; - // break; - // } - // } - // } - // } - // if (!valid) { - // return; - // } + public void send(KafkaProperties.Topic topic, Message message) { ProducerRecord record; if (topic.getPartition() != null) { record = new ProducerRecord(topic.getTopic(), topic.getPartition(), null, message); diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java index d244ce31..feef8038 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java @@ -54,9 +54,9 @@ public class CanalKafkaStarter implements CanalServerStarter { canalKafkaProducer = new CanalKafkaProducer(); canalKafkaProducer.init(kafkaProperties); // set filterTransactionEntry - if (kafkaProperties.isFilterTransactionEntry()) { - System.setProperty("canal.instance.filter.transaction.entry", "true"); - } + // if (kafkaProperties.isFilterTransactionEntry()) { + // System.setProperty("canal.instance.filter.transaction.entry", "true"); + // } // 对应每个instance启动一个worker线程 List destinations = kafkaProperties.getCanalDestinations(); From ebf91397627e22cfbf07263663e40bdb67dbdc4a Mon Sep 17 00:00:00 2001 From: mcy Date: Mon, 3 Sep 2018 17:40:30 +0800 Subject: [PATCH 2/5] =?UTF-8?q?=E5=A2=9E=E5=8A=A0kafka=20producer=20sender?= =?UTF-8?q?=E7=9A=84callback=EF=BC=8Ccanal=E5=92=8Ckafka=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=E6=8F=90=E4=BA=A4=E5=9B=9E=E6=BB=9A=20=E5=8E=BB=E9=99=A4?= =?UTF-8?q?=E4=B8=80=E4=B8=AAcanal=E5=AE=9E=E4=BE=8B=E5=AF=B9=E5=BA=94?= =?UTF-8?q?=E5=A4=9A=E4=B8=AAkafka=20topic=E7=9A=84=E6=9C=BA=E5=88=B6?= =?UTF-8?q?=EF=BC=8C=E4=B8=80=E4=B8=AA=E5=AE=9E=E4=BE=8B=E5=8F=AA=E5=AF=B9?= =?UTF-8?q?=E5=BA=94=E4=B8=80=E4=B8=AAtopic=20kafka=E6=A8=A1=E5=BC=8F?= =?UTF-8?q?=E4=B8=8Bserver=E7=AB=AF=E7=9A=84=E6=89=B9=E6=AC=A1=E5=A4=A7?= =?UTF-8?q?=E5=B0=8F=E4=B8=8D=E8=83=BD=E8=B6=85=E8=BF=871M=EF=BC=8C?= =?UTF-8?q?=E6=9C=80=E5=A4=A7=E4=B8=BA900K?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../loader/CanalAdapterKafkaWorker.java | 6 +-- .../adapter/loader/CanalAdapterLoader.java | 15 +++---- .../adapter/loader/CanalAdapterWorker.java | 13 +++--- deployer/src/main/resources/kafka.yml | 6 +-- .../otter/canal/kafka/CanalKafkaProducer.java | 38 +++++++++++++----- .../otter/canal/kafka/CanalKafkaStarter.java | 40 ++++++++++--------- 6 files changed, 66 insertions(+), 52 deletions(-) diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java index aecaceb2..5cf86611 100644 --- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java +++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java @@ -97,9 +97,9 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { try { logger.info("=============> Start to connect topic: {} <=============", this.topic); connector.connect(); - logger.info("=============> Start to subscribe topic: {}<=============", this.topic); + logger.info("=============> Start to subscribe topic: {} <=============", this.topic); connector.subscribe(); - logger.info("=============> Subscribe topic: {} succeed<=============", this.topic); + logger.info("=============> Subscribe topic: {} succeed <=============", this.topic); while (running) { try { // switcher.get(); //等待开关开启 @@ -146,7 +146,7 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { long currentTS = System.currentTimeMillis(); while (executing.get()) { // 大于1分钟未消费完ack一次keep alive - if (System.currentTimeMillis() - currentTS > 60000) { + if (System.currentTimeMillis() - currentTS > 60000) { connector.ack(); currentTS = System.currentTimeMillis(); } diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java index 98973f11..14efe493 100644 --- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java +++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java @@ -1,12 +1,5 @@ package com.alibaba.otter.canal.client.adapter.loader; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; -import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; -import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; -import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import java.net.InetSocketAddress; import java.net.SocketAddress; import java.util.ArrayList; @@ -16,6 +9,14 @@ import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; +import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; +import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader; + /** * 外部适配器的加载器 * diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java index bca40041..e341c6a3 100644 --- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java +++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java @@ -1,16 +1,15 @@ package com.alibaba.otter.canal.client.adapter.loader; -import com.alibaba.otter.canal.client.CanalConnector; -import com.alibaba.otter.canal.client.CanalConnectors; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; -import com.alibaba.otter.canal.client.adapter.loader.AbstractCanalAdapterWorker; -import com.alibaba.otter.canal.client.impl.ClusterCanalConnector; -import com.alibaba.otter.canal.protocol.Message; - import java.net.SocketAddress; import java.util.List; import java.util.concurrent.Executors; +import com.alibaba.otter.canal.client.CanalConnector; +import com.alibaba.otter.canal.client.CanalConnectors; +import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.impl.ClusterCanalConnector; +import com.alibaba.otter.canal.protocol.Message; + /** * 原生canal-server对应的client适配器工作线程 * diff --git a/deployer/src/main/resources/kafka.yml b/deployer/src/main/resources/kafka.yml index 8abf4934..8f897e6d 100644 --- a/deployer/src/main/resources/kafka.yml +++ b/deployer/src/main/resources/kafka.yml @@ -3,15 +3,11 @@ retries: 0 batchSize: 16384 lingerMs: 1 bufferMemory: 33554432 -filterTransactionEntry: true +canalBatchSize: 50 # Canal的batch size, 默认50K, 由于kafka最大消息体限制请勿超过1M(900K以下) canalDestinations: - canalDestination: example topic: example partition: - # 一个destination可以对应多个topic - #topics: - # - topic: example - # partition: 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 a6d084e4..ae63f415 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,11 @@ package com.alibaba.otter.canal.kafka; -import java.io.IOException; import java.util.Properties; 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.common.serialization.StringSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -48,16 +48,32 @@ public class CanalKafkaProducer { } } - public void send(KafkaProperties.Topic topic, Message message) { - ProducerRecord record; - if (topic.getPartition() != null) { - record = new ProducerRecord(topic.getTopic(), topic.getPartition(), null, message); - } else { - record = new ProducerRecord(topic.getTopic(), message); - } - producer.send(record); - if (logger.isDebugEnabled()) { - logger.debug("send message to kafka topic: {} \n {}", topic.getTopic(), message.toString()); + public void send(KafkaProperties.Topic topic, Message message, Callback callback) { + producer.initTransactions(); + try { + producer.beginTransaction(); + ProducerRecord record; + if (topic.getPartition() != null) { + record = new ProducerRecord(topic.getTopic(), topic.getPartition(), null, message); + } else { + record = new ProducerRecord(topic.getTopic(), message); + } + producer.send(record); + producer.commitTransaction(); + callback.commit(); + if (logger.isDebugEnabled()) { + logger.debug("send message to kafka topic: {} \n {}", topic.getTopic(), message.toString()); + } + } catch (KafkaException e) { + producer.abortTransaction(); + callback.rollback(); } } + + public interface Callback { + + void commit(); + + void rollback(); + } } diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java index feef8038..4ad17e25 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java @@ -33,9 +33,9 @@ public class CanalKafkaStarter implements CanalServerStarter { private ExecutorService executorService; - private CanalKafkaProducer canalKafkaProducer; + private CanalKafkaProducer canalKafkaProducer; - private KafkaProperties kafkaProperties; + private KafkaProperties kafkaProperties; public void init() { try { @@ -55,7 +55,7 @@ public class CanalKafkaStarter implements CanalServerStarter { canalKafkaProducer.init(kafkaProperties); // set filterTransactionEntry // if (kafkaProperties.isFilterTransactionEntry()) { - // System.setProperty("canal.instance.filter.transaction.entry", "true"); + // System.setProperty("canal.instance.filter.transaction.entry", "true"); // } // 对应每个instance启动一个worker线程 List destinations = kafkaProperties.getCanalDestinations(); @@ -101,8 +101,8 @@ public class CanalKafkaStarter implements CanalServerStarter { while (!running) ; logger.info("## start the canal consumer: {}.", destination.getCanalDestination()); - CanalServerWithEmbedded server = CanalServerWithEmbedded.instance(); - ClientIdentity clientIdentity = new ClientIdentity(destination.getCanalDestination(), (short) 1001, ""); + final CanalServerWithEmbedded server = CanalServerWithEmbedded.instance(); + final ClientIdentity clientIdentity = new ClientIdentity(destination.getCanalDestination(), (short) 1001, ""); while (running) { try { if (!server.getCanalInstances().containsKey(clientIdentity.getDestination())) { @@ -118,26 +118,28 @@ public class CanalKafkaStarter implements CanalServerStarter { while (running) { Message message = server.getWithoutAck(clientIdentity, kafkaProperties.getCanalBatchSize()); // 获取指定数量的数据 - long batchId = message.getId(); + final long batchId = message.getId(); try { int size = message.isRaw() ? message.getRawEntries().size() : message.getEntries().size(); if (batchId != -1 && size != 0) { - if (!StringUtils.isEmpty(destination.getTopic())) { - Topic topic = new Topic(); - topic.setTopic(destination.getTopic()); - topic.setPartition(destination.getPartition()); - destination.getTopics().add(topic); - } - for (Topic topic : destination.getTopics()) { - canalKafkaProducer.send(topic, message); // 发送message到所有topic - } + Topic topic = new Topic(); + topic.setTopic(destination.getTopic()); + topic.setPartition(destination.getPartition()); + canalKafkaProducer.send(topic, message, new CanalKafkaProducer.Callback() { + + @Override + public void commit() { + server.ack(clientIdentity, batchId); // 提交确认 + } + + @Override + public void rollback() { + server.rollback(clientIdentity, batchId); + } + }); // 发送message到topic } - if (batchId != -1) { - server.ack(clientIdentity, batchId); // 提交确认 - } } catch (Exception e) { - server.rollback(clientIdentity); logger.error(e.getMessage(), e); } } From 17f345bfe5bdf7d7eb00ffe6667573d7ec5bab84 Mon Sep 17 00:00:00 2001 From: mcy Date: Mon, 3 Sep 2018 18:07:43 +0800 Subject: [PATCH 3/5] =?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(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(); } } From 81c216f1c16a113047202b188a158074c0797612 Mon Sep 17 00:00:00 2001 From: mcy Date: Tue, 4 Sep 2018 09:39:02 +0800 Subject: [PATCH 4/5] =?UTF-8?q?=E5=A2=9E=E5=8A=A0kafka=20get=E7=9A=84?= =?UTF-8?q?=E8=B6=85=E6=97=B6=E6=97=B6=E9=97=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- deployer/src/main/resources/kafka.yml | 6 +++++- .../alibaba/otter/canal/kafka/CanalKafkaStarter.java | 12 +++++++++++- .../alibaba/otter/canal/kafka/KafkaProperties.java | 9 +++++++++ 3 files changed, 25 insertions(+), 2 deletions(-) diff --git a/deployer/src/main/resources/kafka.yml b/deployer/src/main/resources/kafka.yml index 8f897e6d..edd6b464 100644 --- a/deployer/src/main/resources/kafka.yml +++ b/deployer/src/main/resources/kafka.yml @@ -3,7 +3,11 @@ retries: 0 batchSize: 16384 lingerMs: 1 bufferMemory: 33554432 -canalBatchSize: 50 # Canal的batch size, 默认50K, 由于kafka最大消息体限制请勿超过1M(900K以下) + +# Canal的batch size, 默认50K, 由于kafka最大消息体限制请勿超过1M(900K以下) +canalBatchSize: 50 +# Canal get数据的超时时间, 单位: 毫秒, 空为不限超时 +canalGetTimeout: 100 canalDestinations: - canalDestination: example diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java index 4ad17e25..717a4525 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java @@ -4,6 +4,7 @@ import java.io.FileInputStream; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; @@ -117,7 +118,16 @@ public class CanalKafkaStarter implements CanalServerStarter { logger.info("## the canal consumer {} is running now ......", destination.getCanalDestination()); while (running) { - Message message = server.getWithoutAck(clientIdentity, kafkaProperties.getCanalBatchSize()); // 获取指定数量的数据 + Message message; + if (kafkaProperties.getCanalGetTimeout() != null) { + message = server.getWithoutAck(clientIdentity, + kafkaProperties.getCanalBatchSize(), + kafkaProperties.getCanalGetTimeout(), + TimeUnit.MILLISECONDS); + } else { + message = server.getWithoutAck(clientIdentity, kafkaProperties.getCanalBatchSize()); + } + final long batchId = message.getId(); try { int size = message.isRaw() ? message.getRawEntries().size() : message.getEntries().size(); diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java b/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java index 9194c0ab..40f11bf5 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java @@ -20,6 +20,7 @@ public class KafkaProperties { private long bufferMemory = 33554432L; private boolean filterTransactionEntry = true; private int canalBatchSize = 50; + private Long canalGetTimeout; private List canalDestinations = new ArrayList(); @@ -151,6 +152,14 @@ public class KafkaProperties { this.canalBatchSize = canalBatchSize; } + public Long getCanalGetTimeout() { + return canalGetTimeout; + } + + public void setCanalGetTimeout(Long canalGetTimeout) { + this.canalGetTimeout = canalGetTimeout; + } + public List getCanalDestinations() { return canalDestinations; } From 728550cac3dd9a400f2d0077c363a25b4a3a399b Mon Sep 17 00:00:00 2001 From: mcy Date: Tue, 4 Sep 2018 09:56:51 +0800 Subject: [PATCH 5/5] =?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(); }