From ad62275cd73971f2925c21bf3be51accec1459b1 Mon Sep 17 00:00:00 2001 From: mcy Date: Tue, 5 Feb 2019 19:20:36 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0kafka=E8=87=AA=E5=AE=9A?= =?UTF-8?q?=E4=B9=89properties?= 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 | 12 +++++- deployer/src/main/resources/canal.properties | 3 +- .../otter/canal/common/MQProperties.java | 43 ++++++++++++------- .../otter/canal/kafka/CanalKafkaProducer.java | 7 ++- 5 files changed, 47 insertions(+), 19 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 440d8091..f387e1b4 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 @@ -51,6 +51,7 @@ public class CanalConstants { 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"; + public static final String CANAL_MQ_PROPERTIES = ROOT + "." + "mq.properties"; public static String getInstanceModeKey(String destination) { return MessageFormat.format(INSTANCE_MODE_TEMPLATE, destination); 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 19774812..05c8c6f0 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 @@ -1,5 +1,6 @@ package com.alibaba.otter.canal.deployer; +import java.util.Map; import java.util.Properties; import org.apache.commons.lang.StringUtils; @@ -162,7 +163,16 @@ public class CanalStater { if (!StringUtils.isEmpty(transaction)) { mqProperties.setTransaction(Boolean.valueOf(transaction)); } + + for (Object key : properties.keySet()) { + key = StringUtils.trim(key.toString()); + if (((String) key).startsWith(CanalConstants.CANAL_MQ_PROPERTIES)) { + String value = CanalController.getProperty(properties, (String) key); + String subKey = ((String) key).substring(CanalConstants.CANAL_MQ_PROPERTIES.length() + 1); + mqProperties.getProperties().put(subKey, value); + } + } + return mqProperties; } - } diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index fcdf6ac2..8b5c09fc 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -116,4 +116,5 @@ canal.mq.flatMessage = true canal.mq.compressionType = none canal.mq.acks = all # use transaction for kafka flatMessage batch produce -canal.mq.transaction = false \ No newline at end of file +canal.mq.transaction = false +#canal.mq.properties. = \ 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 ccbf4f5e..4ee28c5e 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 @@ -1,5 +1,7 @@ package com.alibaba.otter.canal.common; +import java.util.Properties; + /** * kafka 配置项 * @@ -8,22 +10,23 @@ package com.alibaba.otter.canal.common; */ public class MQProperties { - private String servers = "127.0.0.1:6667"; - private int retries = 0; - private int batchSize = 16384; - private int lingerMs = 1; - private int maxRequestSize = 1048576; - private long bufferMemory = 33554432L; - private boolean filterTransactionEntry = true; - private String producerGroup = "Canal-Producer"; - private int canalBatchSize = 50; - private Long canalGetTimeout = 100L; - private boolean flatMessage = true; - private String compressionType = "none"; - private String acks = "all"; - private String aliyunAccessKey = ""; - private String aliyunSecretKey = ""; - private boolean transaction = false; // 是否开启事务 + private String servers = "127.0.0.1:6667"; + private int retries = 0; + private int batchSize = 16384; + private int lingerMs = 1; + private int maxRequestSize = 1048576; + private long bufferMemory = 33554432L; + private boolean filterTransactionEntry = true; + private String producerGroup = "Canal-Producer"; + private int canalBatchSize = 50; + private Long canalGetTimeout = 100L; + private boolean flatMessage = true; + private String compressionType = "none"; + private String acks = "all"; + private String aliyunAccessKey = ""; + private String aliyunSecretKey = ""; + private boolean transaction = false; // 是否开启事务 + private Properties properties = new Properties(); public static class CanalDestination { @@ -210,4 +213,12 @@ public class MQProperties { public void setTransaction(boolean transaction) { this.transaction = transaction; } + + public Properties getProperties() { + return properties; + } + + public void setProperties(Properties properties) { + this.properties = properties; + } } 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 55133264..12c048aa 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 @@ -49,7 +49,12 @@ public class CanalKafkaProducer implements CanalMQProducer { properties.put("max.request.size", kafkaProperties.getMaxRequestSize()); properties.put("buffer.memory", kafkaProperties.getBufferMemory()); properties.put("key.serializer", StringSerializer.class.getName()); - if(kafkaProperties.getTransaction()){ + + if (!kafkaProperties.getProperties().isEmpty()) { + properties.putAll(kafkaProperties.getProperties()); + } + + if (kafkaProperties.getTransaction()) { properties.put("transactional.id", "canal-transactional-id"); } else { properties.put("retries", kafkaProperties.getRetries());