From 2f214ce3a65dabcce5d4e89f7bee1573ecf6a739 Mon Sep 17 00:00:00 2001 From: machey Date: Sun, 20 Jan 2019 14:20:06 +0800 Subject: [PATCH 1/4] =?UTF-8?q?=E9=87=87=E7=94=A8=E4=BA=8B=E5=8A=A1?= =?UTF-8?q?=E6=89=B9=E9=87=8F=E5=BC=82=E6=AD=A5=E6=8A=95=E9=80=92kafka=20f?= =?UTF-8?q?latMessage=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, From 4ebf17f57b8c278809f081b9e59d2c89e2a6b4ef Mon Sep 17 00:00:00 2001 From: machey Date: Sun, 20 Jan 2019 14:30:20 +0800 Subject: [PATCH 2/4] =?UTF-8?q?=E4=BB=A3=E7=A0=81=E8=B0=83=E6=95=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/kafka/CanalKafkaProducer.java | 58 ++++++++----------- 1 file changed, 25 insertions(+), 33 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 cf5931ef..1177a82e 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 @@ -140,11 +140,12 @@ public class CanalKafkaProducer implements CanalMQProducer { if (flatMessages != null) { // 开启事务，需要kafka版本支持 if (kafkaProperties.getTransaction()) { + producer.initTransactions(); producer.beginTransaction(); } - for (FlatMessage flatMessage : flatMessages) { - if (StringUtils.isEmpty(canalDestination.getPartitionHash())) { - try { + try { + for (FlatMessage flatMessage : flatMessages) { + if (StringUtils.isEmpty(canalDestination.getPartitionHash())) { Integer partition = canalDestination.getPartition(); if (partition == null) { partition = 0; @@ -158,21 +159,14 @@ public class CanalKafkaProducer implements CanalMQProducer { } else { producer2.send(record).get(); } - } catch (Exception e) { - if (kafkaProperties.getTransaction()) { - producer.abortTransaction(); - } - throw e; - } - } else { - FlatMessage[] partitionFlatMessage = MQMessageUtils.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 { + } else { + FlatMessage[] partitionFlatMessage = MQMessageUtils.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( topicName, i, @@ -183,30 +177,28 @@ public class CanalKafkaProducer implements CanalMQProducer { } else { producer2.send(record).get(); } - } catch (Exception e) { - 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, + JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue)); + } + } + } catch (Exception e) { if (kafkaProperties.getTransaction()) { - producer.commitTransaction(); - } - - if (logger.isDebugEnabled()) { - logger.debug("Send flat message to kafka topic: [{}], packet: {}", - topicName, - JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue)); + producer.abortTransaction(); } + throw e; } } } - - // producer.commitTransaction(); } } From 851637611f43765bf076a42b24654aad90893373 Mon Sep 17 00:00:00 2001 From: machey Date: Sun, 20 Jan 2019 15:02:06 +0800 Subject: [PATCH 3/4] modify --- .../otter/canal/kafka/CanalKafkaProducer.java | 122 +++++++++--------- 1 file changed, 64 insertions(+), 58 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 1177a82e..a7e65d51 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 @@ -3,6 +3,7 @@ package com.alibaba.otter.canal.kafka; import java.util.List; import java.util.Map; import java.util.Properties; +import java.util.concurrent.ExecutionException; import org.apache.commons.lang.StringUtils; import org.apache.kafka.clients.producer.KafkaProducer; @@ -56,8 +57,13 @@ public class CanalKafkaProducer implements CanalMQProducer { properties.put("value.serializer", StringSerializer.class.getName()); producer2 = new KafkaProducer(properties); } - - // producer.initTransactions(); + if (kafkaProperties.getTransaction()) { + if (!kafkaProperties.getFlatMessage()) { + producer.initTransactions(); + } else { + producer2.initTransactions(); + } + } } @Override @@ -79,6 +85,17 @@ public class CanalKafkaProducer implements CanalMQProducer { @Override public void send(MQProperties.CanalDestination canalDestination, Message message, Callback callback) { + // 开启事务，需要kafka版本支持 + Producer producerTmp; + if (!kafkaProperties.getFlatMessage()) { + producerTmp = producer; + } else { + producerTmp = producer2; + } + + if (kafkaProperties.getTransaction()) { + producerTmp.beginTransaction(); + } try { if (!StringUtils.isEmpty(canalDestination.getDynamicTopic())) { // 动态topic @@ -96,9 +113,15 @@ public class CanalKafkaProducer implements CanalMQProducer { } else { send(canalDestination, canalDestination.getTopic(), message); } + if (kafkaProperties.getTransaction()) { + producerTmp.commitTransaction(); + } callback.commit(); } catch (Exception e) { logger.error(e.getMessage(), e); + if (kafkaProperties.getTransaction()) { + producerTmp.abortTransaction(); + } callback.rollback(); } } @@ -127,8 +150,11 @@ public class CanalKafkaProducer implements CanalMQProducer { } if (record != null) { - // 同步发送原生message - producer.send(record).get(); + if (kafkaProperties.getTransaction()) { + producer.send(record); + } else { + producer.send(record).get(); + } if (logger.isDebugEnabled()) { logger.debug("Send message to kafka topic: [{}], packet: {}", topicName, message.toString()); @@ -138,67 +164,47 @@ public class CanalKafkaProducer implements CanalMQProducer { // 发送扁平数据json List flatMessages = MQMessageUtils.messageConverter(message); if (flatMessages != null) { - // 开启事务，需要kafka版本支持 - if (kafkaProperties.getTransaction()) { - producer.initTransactions(); - producer.beginTransaction(); - } - try { - for (FlatMessage flatMessage : flatMessages) { - if (StringUtils.isEmpty(canalDestination.getPartitionHash())) { - Integer partition = canalDestination.getPartition(); - if (partition == null) { - partition = 0; - } - ProducerRecord record = new ProducerRecord(topicName, - partition, - null, - JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue)); - if (kafkaProperties.getTransaction()) { - producer2.send(record); - } else { - producer2.send(record).get(); - } - } else { - FlatMessage[] partitionFlatMessage = MQMessageUtils.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( - topicName, - i, - null, - JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue)); - if (kafkaProperties.getTransaction()) { - producer2.send(record); - } else { - producer2.send(record).get(); - } - } - } + for (FlatMessage flatMessage : flatMessages) { + if (StringUtils.isEmpty(canalDestination.getPartitionHash())) { + Integer partition = canalDestination.getPartition(); + if (partition == null) { + partition = 0; } - - if (kafkaProperties.getTransaction()) { - producer.commitTransaction(); - } - - if (logger.isDebugEnabled()) { - logger.debug("Send flat message to kafka topic: [{}], packet: {}", - topicName, - JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue)); + produce(topicName, partition, flatMessage); + } else { + FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage, + canalDestination.getPartitionsNum(), + canalDestination.getPartitionHash()); + int length = partitionFlatMessage.length; + for (int i = 0; i < length; i++) { + FlatMessage flatMessagePart = partitionFlatMessage[i]; + if (flatMessagePart != null) { + produce(topicName, i, flatMessagePart); + } } } - } catch (Exception e) { - if (kafkaProperties.getTransaction()) { - producer.abortTransaction(); + + if (logger.isDebugEnabled()) { + logger.debug("Send flat message to kafka topic: [{}], packet: {}", + topicName, + JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue)); } - throw e; } } } } + private void produce(String topicName, int partition, FlatMessage flatMessage) throws ExecutionException, + InterruptedException { + ProducerRecord record = new ProducerRecord(topicName, + partition, + null, + JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue)); + if (kafkaProperties.getTransaction()) { + producer2.send(record); + } else { + producer2.send(record).get(); + } + } + } From 197ca67c8bb798fd99f1fba1beb166a2bef8e1d6 Mon Sep 17 00:00:00 2001 From: mcy Date: Mon, 21 Jan 2019 09:56:35 +0800 Subject: [PATCH 4/4] =?UTF-8?q?=E6=B5=8B=E8=AF=95=E8=B0=83=E6=95=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../com/alibaba/otter/canal/kafka/CanalKafkaProducer.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) 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 a7e65d51..55133264 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 @@ -44,12 +44,16 @@ public class CanalKafkaProducer implements CanalMQProducer { properties.put("bootstrap.servers", kafkaProperties.getServers()); properties.put("acks", kafkaProperties.getAcks()); properties.put("compression.type", kafkaProperties.getCompressionType()); - properties.put("retries", kafkaProperties.getRetries()); properties.put("batch.size", kafkaProperties.getBatchSize()); properties.put("linger.ms", kafkaProperties.getLingerMs()); properties.put("max.request.size", kafkaProperties.getMaxRequestSize()); properties.put("buffer.memory", kafkaProperties.getBufferMemory()); properties.put("key.serializer", StringSerializer.class.getName()); + if(kafkaProperties.getTransaction()){ + properties.put("transactional.id", "canal-transactional-id"); + } else { + properties.put("retries", kafkaProperties.getRetries()); + } if (!kafkaProperties.getFlatMessage()) { properties.put("value.serializer", MessageSerializer.class.getName()); producer = new KafkaProducer(properties);