diff --git a/client-launcher/pom.xml b/client-launcher/pom.xml
index 623f2a97..8c29047a 100644
--- a/client-launcher/pom.xml
+++ b/client-launcher/pom.xml
@@ -23,6 +23,12 @@
canal.client
${project.version}
+
+
+ org.apache.rocketmq
+ rocketmq-client
+ 4.3.0
+
org.apache.kafka
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 72320bb5..4429ca80 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
@@ -96,7 +96,7 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
while (!running)
;
ExecutorService executor = Executors.newSingleThreadExecutor();
- final AtomicBoolean executing = new AtomicBoolean(true);
+ // final AtomicBoolean executing = new AtomicBoolean(true);
while (running) {
try {
logger.info("=============> Start to connect topic: {} <=============", this.topic);
@@ -116,36 +116,41 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
}
if (messages != null) {
for (final Object message : messages) {
- executing.set(true);
- if (message != null) {
- executor.submit(new Runnable() {
-
- @Override
- public void run() {
- try {
- if (message instanceof FlatMessage) {
- writeOut((FlatMessage) message);
- } else {
- writeOut((Message) message);
- }
- } catch (Exception e) {
- logger.error(e.getMessage(), e);
- } finally {
- executing.compareAndSet(true, false);
- }
- }
- });
-
- // 间隔一段时间ack一次, 防止因超时未响应切换到另外台客户端
- long currentTS = System.currentTimeMillis();
- while (executing.get()) {
- // 大于10秒未消费完ack一次keep alive
- if (System.currentTimeMillis() - currentTS > 10000) {
- connector.ack();
- currentTS = System.currentTimeMillis();
- }
- }
+ if (message instanceof FlatMessage) {
+ writeOut((FlatMessage) message);
+ } else {
+ writeOut((Message) message);
}
+ // executing.set(true);
+ // if (message != null) {
+ // executor.submit(new Runnable() {
+ //
+ // @Override
+ // public void run() {
+ // try {
+ // if (message instanceof FlatMessage) {
+ // writeOut((FlatMessage) message);
+ // } else {
+ // writeOut((Message) message);
+ // }
+ // } catch (Exception e) {
+ // logger.error(e.getMessage(), e);
+ // } finally {
+ // executing.compareAndSet(true, false);
+ // }
+ // }
+ // });
+ //
+ // // 间隔一段时间ack一次, 防止因超时未响应切换到另外台客户端
+ // long currentTS = System.currentTimeMillis();
+ // while (executing.get()) {
+ // // 大于10秒未消费完ack一次keep alive
+ // if (System.currentTimeMillis() - currentTS > 10000) {
+ // connector.ack();
+ // currentTS = System.currentTimeMillis();
+ // }
+ // }
+ // }
}
}
connector.ack();
diff --git a/client/pom.xml b/client/pom.xml
index 43b3067f..370d1854 100644
--- a/client/pom.xml
+++ b/client/pom.xml
@@ -101,12 +101,14 @@
${spring_version}
test
+
org.apache.rocketmq
rocketmq-client
4.3.0
+ provided
-
+
org.apache.kafka
kafka-clients
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 0a3a0dd6..b3f8800e 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
@@ -233,7 +233,7 @@ public class KafkaCanalConnector {
kafkaConsumer.commitSync();
}
if (kafkaConsumer2 != null) {
- kafkaConsumer2.commitAsync();
+ kafkaConsumer2.commitSync();
}
}
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 63827b15..15217f13 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,14 @@
package com.alibaba.otter.canal.kafka;
+import java.util.ArrayList;
import java.util.List;
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.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -14,8 +17,7 @@ import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.common.MQProperties;
import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.protocol.Message;
-import com.alibaba.otter.canal.spi.CanalMQProducer;
-;
+import com.alibaba.otter.canal.spi.CanalMQProducer;;
/**
* kafka producer 主操作类
@@ -74,9 +76,10 @@ public class CanalKafkaProducer implements CanalMQProducer {
@Override
public void send(MQProperties.CanalDestination canalDestination, Message message, Callback callback) {
- try {
- // producer.beginTransaction();
- if (!kafkaProperties.getFlatMessage()) {
+
+ // producer.beginTransaction();
+ if (!kafkaProperties.getFlatMessage()) {
+ try {
ProducerRecord record;
if (canalDestination.getPartition() != null) {
record = new ProducerRecord(canalDestination.getTopic(),
@@ -87,58 +90,76 @@ public class CanalKafkaProducer implements CanalMQProducer {
record = new ProducerRecord(canalDestination.getTopic(), 0, null, message);
}
- producer.send(record);
- } else {
- // 发送扁平数据json
- List flatMessages = FlatMessage.messageConverter(message);
- if (flatMessages != null) {
- for (FlatMessage flatMessage : flatMessages) {
- if (canalDestination.getPartition() != null) {
- ProducerRecord record = new ProducerRecord(canalDestination.getTopic(),
- canalDestination.getPartition(),
- null,
- JSON.toJSONString(flatMessage));
+ producer.send(record).get();
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ // producer.abortTransaction();
+ callback.rollback();
+ }
+ } else {
+ // 发送扁平数据json
+ List flatMessages = FlatMessage.messageConverter(message);
+ if (flatMessages != null) {
+ for (FlatMessage flatMessage : flatMessages) {
+ if (canalDestination.getPartition() != null) {
+ try {
+ ProducerRecord record = new ProducerRecord(canalDestination
+ .getTopic(), canalDestination.getPartition(), null, JSON.toJSONString(flatMessage));
producer2.send(record);
- } else {
- if (canalDestination.getPartitionHash() != null
- && !canalDestination.getPartitionHash().isEmpty()) {
- FlatMessage[] partitionFlatMessage = FlatMessage.messagePartition(flatMessage,
- canalDestination.getPartitionsNum(),
- canalDestination.getPartitionHash());
- int length = partitionFlatMessage.length;
- for (int i = 0; i < length; i++) {
- FlatMessage flatMessagePart = partitionFlatMessage[i];
- if (flatMessagePart != null) {
- ProducerRecord record = new ProducerRecord(canalDestination.getTopic(),
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ // producer.abortTransaction();
+ callback.rollback();
+ }
+ } else {
+ if (canalDestination.getPartitionHash() != null
+ && !canalDestination.getPartitionHash().isEmpty()) {
+ FlatMessage[] partitionFlatMessage = FlatMessage.messagePartition(flatMessage,
+ canalDestination.getPartitionsNum(),
+ canalDestination.getPartitionHash());
+ int length = partitionFlatMessage.length;
+ for (int i = 0; i < length; i++) {
+ FlatMessage flatMessagePart = partitionFlatMessage[i];
+ if (flatMessagePart != null) {
+ try {
+ ProducerRecord record = new ProducerRecord(
+ canalDestination.getTopic(),
i,
null,
JSON.toJSONString(flatMessagePart));
- producer2.send(record);
+ producer2.send(record).get();
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ // producer.abortTransaction();
+ callback.rollback();
}
}
- } else {
- ProducerRecord record = new ProducerRecord(canalDestination.getTopic(),
+ }
+ } else {
+ try {
+ ProducerRecord record = new ProducerRecord(
+ canalDestination.getTopic(),
0,
null,
JSON.toJSONString(flatMessage));
- producer2.send(record);
+ producer2.send(record).get();
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ // producer.abortTransaction();
+ callback.rollback();
}
}
-
}
}
}
-
- // producer.commitTransaction();
- callback.commit();
- if (logger.isDebugEnabled()) {
- logger.debug("send message to kafka topic: {}", canalDestination.getTopic());
- }
- } catch (Exception e) {
- logger.error(e.getMessage(), e);
- // producer.abortTransaction();
- callback.rollback();
}
+
+ // producer.commitTransaction();
+ callback.commit();
+ if (logger.isDebugEnabled()) {
+ logger.debug("send message to kafka topic: {}", canalDestination.getTopic());
+ }
+
}
}