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/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()); + } + } }