From fea4d8210a52e2004e4988e05192a8dd38f66937 Mon Sep 17 00:00:00 2001
From: "jianghang.loujh"
@@ -27,20 +27,20 @@ public class PulsarMQProducerConfig extends MQProperties { * 在发送消息时会自动拼接上 *
*/ - private String topicTenantPrefix; + private String topicTenantPrefix; /** * 生产者角色权限,请确保该角色有canal使用的所有topic生产者权限(最低要求) */ - private String roleToken; + private String roleToken; /** * admin服务器地址 */ - private String adminServerUrl; + private String adminServerUrl; /** * listener name */ - private String listenerName; + private String listenerName; /** * enableChunking @@ -50,7 +50,7 @@ public class PulsarMQProducerConfig extends MQProperties { /** * compressionType */ - private String compressionType; + private String compressionType; public String getServerUrl() { return serverUrl; @@ -95,6 +95,7 @@ public class PulsarMQProducerConfig extends MQProperties { public void setEnableChunking(boolean enableChunking) { this.enableChunking = enableChunking; } + public boolean getEnableChunking() { return this.enableChunking; } @@ -102,6 +103,7 @@ public class PulsarMQProducerConfig extends MQProperties { public void setCompressionType(String compressionType) { this.compressionType = compressionType; } + public String getCompressionType() { return this.compressionType; } diff --git a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/producer/CanalPulsarMQProducer.java b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/producer/CanalPulsarMQProducer.java index 38ca7078..4f4d581d 100644 --- a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/producer/CanalPulsarMQProducer.java +++ b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/producer/CanalPulsarMQProducer.java @@ -147,7 +147,6 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ tmpProperties.setCompressionType(compressionType); } - if (logger.isDebugEnabled()) { logger.debug("Load pulsar properties ==> {}", JSON.toJSON(this.mqProperties)); } @@ -420,13 +419,13 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ // 创建指定topic的生产者 ProducerBuilder producerBuilder = client.newProducer(); - if(pulsarMQProperties.getEnableChunking()){ + if (pulsarMQProperties.getEnableChunking()) { producerBuilder.enableChunking(true); producerBuilder.enableBatching(false); } - if(!StringUtils.isEmpty(pulsarMQProperties.getCompressionType())) { - switch(pulsarMQProperties.getCompressionType().toLowerCase()) { + if (!StringUtils.isEmpty(pulsarMQProperties.getCompressionType())) { + switch (pulsarMQProperties.getCompressionType().toLowerCase()) { case "lz4": producerBuilder.compressionType(CompressionType.LZ4); break;