From 2f214ce3a65dabcce5d4e89f7bee1573ecf6a739 Mon Sep 17 00:00:00 2001 From: machey Date: Sun, 20 Jan 2019 14:20:06 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=87=E7=94=A8=E4=BA=8B=E5=8A=A1=E6=89=B9?= =?UTF-8?q?=E9=87=8F=E5=BC=82=E6=AD=A5=E6=8A=95=E9=80=92kafka=20flatMessag?= =?UTF-8?q?e=20fixed=20#1454?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/deployer/CanalConstants.java | 1 + .../otter/canal/deployer/CanalStater.java | 5 +- deployer/src/main/resources/canal.properties | 4 +- .../otter/canal/common/MQProperties.java | 9 +++ .../otter/canal/kafka/CanalKafkaProducer.java | 76 ++++++++++--------- 5 files changed, 56 insertions(+), 39 deletions(-) diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java index ba39bde0..440d8091 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java @@ -48,6 +48,7 @@ public class CanalConstants { public static final String CANAL_MQ_FLATMESSAGE = ROOT + "." + "mq.flatMessage"; public static final String CANAL_MQ_COMPRESSION_TYPE = ROOT + "." + "mq.compressionType"; public static final String CANAL_MQ_ACKS = ROOT + "." + "mq.acks"; + public static final String CANAL_MQ_TRANSACTION = ROOT + "." + "mq.transaction"; public static final String CANAL_ALIYUN_ACCESSKEY = ROOT + "." + "aliyun.accessKey"; public static final String CANAL_ALIYUN_SECRETKEY = ROOT + "." + "aliyun.secretKey"; diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java index f8503fa9..19774812 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java @@ -150,7 +150,6 @@ public class CanalStater { if (!StringUtils.isEmpty(acks)) { mqProperties.setAcks(acks); } - String aliyunAccessKey = CanalController.getProperty(properties, CanalConstants.CANAL_ALIYUN_ACCESSKEY); if (!StringUtils.isEmpty(aliyunAccessKey)) { mqProperties.setAliyunAccessKey(aliyunAccessKey); @@ -159,6 +158,10 @@ public class CanalStater { if (!StringUtils.isEmpty(aliyunSecretKey)) { mqProperties.setAliyunSecretKey(aliyunSecretKey); } + String transaction = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_TRANSACTION); + if (!StringUtils.isEmpty(transaction)) { + mqProperties.setTransaction(Boolean.valueOf(transaction)); + } return mqProperties; } diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index a7221726..fcdf6ac2 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -114,4 +114,6 @@ canal.mq.canalBatchSize = 50 canal.mq.canalGetTimeout = 100 canal.mq.flatMessage = true canal.mq.compressionType = none -canal.mq.acks = all \ No newline at end of file +canal.mq.acks = all +# use transaction for kafka flatMessage batch produce +canal.mq.transaction = false \ No newline at end of file diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java index 383ed6bd..ccbf4f5e 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java @@ -23,6 +23,7 @@ public class MQProperties { private String acks = "all"; private String aliyunAccessKey = ""; private String aliyunSecretKey = ""; + private boolean transaction = false; // 是否开启事务 public static class CanalDestination { @@ -201,4 +202,12 @@ public class MQProperties { public void setMaxRequestSize(int maxRequestSize) { this.maxRequestSize = maxRequestSize; } + + public boolean getTransaction() { + return transaction; + } + + public void setTransaction(boolean transaction) { + this.transaction = transaction; + } } 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 490d50b5..cf5931ef 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 @@ -98,56 +98,51 @@ public class CanalKafkaProducer implements CanalMQProducer { } callback.commit(); } catch (Exception e) { + logger.error(e.getMessage(), e); callback.rollback(); } } private void send(MQProperties.CanalDestination canalDestination, String topicName, Message message) throws Exception { - // producer.beginTransaction(); if (!kafkaProperties.getFlatMessage()) { - try { - ProducerRecord record = null; - if (canalDestination.getPartition() != null) { - record = new ProducerRecord<>(topicName, canalDestination.getPartition(), null, message); - } else { - if (canalDestination.getPartitionHash() != null && !canalDestination.getPartitionHash().isEmpty()) { - Message[] messages = MQMessageUtils.messagePartition(message, - canalDestination.getPartitionsNum(), - canalDestination.getPartitionHash()); - int length = messages.length; - for (int i = 0; i < length; i++) { - Message messagePartition = messages[i]; - if (messagePartition != null) { - record = new ProducerRecord<>(topicName, i, null, messagePartition); - } + ProducerRecord record = null; + if (canalDestination.getPartition() != null) { + record = new ProducerRecord<>(topicName, canalDestination.getPartition(), null, message); + } else { + if (canalDestination.getPartitionHash() != null && !canalDestination.getPartitionHash().isEmpty()) { + Message[] messages = MQMessageUtils.messagePartition(message, + canalDestination.getPartitionsNum(), + canalDestination.getPartitionHash()); + int length = messages.length; + for (int i = 0; i < length; i++) { + Message messagePartition = messages[i]; + if (messagePartition != null) { + record = new ProducerRecord<>(topicName, i, null, messagePartition); } - } else { - record = new ProducerRecord<>(topicName, 0, null, message); } + } else { + record = new ProducerRecord<>(topicName, 0, null, message); } + } - if (record != null) { - // 同步发送原生message - producer.send(record).get(); + if (record != null) { + // 同步发送原生message + producer.send(record).get(); - if (logger.isDebugEnabled()) { - logger.debug("Send message to kafka topic: [{}], packet: {}", topicName, message.toString()); - } + if (logger.isDebugEnabled()) { + logger.debug("Send message to kafka topic: [{}], packet: {}", topicName, message.toString()); } - } catch (Exception e) { - logger.error(e.getMessage(), e); - // producer.abortTransaction(); - throw e; } } else { // 发送扁平数据json List flatMessages = MQMessageUtils.messageConverter(message); if (flatMessages != null) { - int idx = 0; - int size = flatMessages.size(); + // 开启事务，需要kafka版本支持 + if (kafkaProperties.getTransaction()) { + producer.beginTransaction(); + } for (FlatMessage flatMessage : flatMessages) { - idx++; if (StringUtils.isEmpty(canalDestination.getPartitionHash())) { try { Integer partition = canalDestination.getPartition(); @@ -158,14 +153,15 @@ public class CanalKafkaProducer implements CanalMQProducer { partition, null, JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue)); - if (idx != size) { + if (kafkaProperties.getTransaction()) { producer2.send(record); } else { producer2.send(record).get(); } } catch (Exception e) { - logger.error(e.getMessage(), e); - // producer.abortTransaction(); + if (kafkaProperties.getTransaction()) { + producer.abortTransaction(); + } throw e; } } else { @@ -182,19 +178,25 @@ public class CanalKafkaProducer implements CanalMQProducer { i, null, JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue)); - if (idx != size) { + if (kafkaProperties.getTransaction()) { producer2.send(record); } else { producer2.send(record).get(); } } catch (Exception e) { - logger.error(e.getMessage(), e); - // producer.abortTransaction(); + if (kafkaProperties.getTransaction()) { + producer.abortTransaction(); + } throw e; } } } } + + if (kafkaProperties.getTransaction()) { + producer.commitTransaction(); + } + if (logger.isDebugEnabled()) { logger.debug("Send flat message to kafka topic: [{}], packet: {}", topicName,