diff --git a/admin/pom.xml b/admin/pom.xml index 94469ab4..4ae240d4 100644 --- a/admin/pom.xml +++ b/admin/pom.xml @@ -36,7 +36,7 @@ org.springframework.boot spring-boot-dependencies - 2.5.4 + 2.5.15 pom import @@ -48,7 +48,12 @@ io.ebean ebean - 11.41.1 + 11.45.1 + + + org.yaml + snakeyaml + 2.0 diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml index 633269b7..4efdedfd 100644 --- a/client-adapter/launcher/pom.xml +++ b/client-adapter/launcher/pom.xml @@ -17,7 +17,7 @@ org.springframework.boot spring-boot-dependencies - 2.5.4 + 2.5.15 pom import diff --git a/client-adapter/pom.xml b/client-adapter/pom.xml index cfd7c4aa..21d023c5 100644 --- a/client-adapter/pom.xml +++ b/client-adapter/pom.xml @@ -117,7 +117,7 @@ org.springframework.boot spring-boot - 2.5.4 + 2.5.15 com.h2database @@ -238,7 +238,7 @@ org.yaml snakeyaml - 1.29 + 2.0 diff --git a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQConstants.java b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQConstants.java index 5a6d1774..bea661f5 100644 --- a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQConstants.java +++ b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQConstants.java @@ -62,8 +62,18 @@ public class PulsarMQConstants { public static final String PULSARMQ_ADMIN_SERVER_URL = ROOT + "." + "adminServerUrl"; /** - * Pulsar admin服务器地址 + * Pulsar 监听器名字 */ public static final String PULSARMQ_LISTENER_NAME = ROOT + "." + "listenerName"; + + /** + * Pulsar 开启chunking + */ + public static final String PULSARMQ_ENABLE_CHUNKING = ROOT + "." + "enableChunking"; + + /** + * Pulsar 压缩算法 + */ + public static final String PULSARMQ_COMPRESSION_TYPE = ROOT + "." + "compressionType"; } diff --git a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQProducerConfig.java b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQProducerConfig.java index 40d29f3d..af51c13f 100644 --- a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQProducerConfig.java +++ b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQProducerConfig.java @@ -42,6 +42,16 @@ public class PulsarMQProducerConfig extends MQProperties { */ private String listenerName; + /** + * enableChunking + */ + private boolean enableChunking; + + /** + * compressionType + */ + private String compressionType; + public String getServerUrl() { return serverUrl; } @@ -81,4 +91,18 @@ public class PulsarMQProducerConfig extends MQProperties { public void setListenerName(String listenerName) { this.listenerName = listenerName; } + + public void setEnableChunking(boolean enableChunking) { + this.enableChunking = enableChunking; + } + public boolean getEnableChunking() { + return this.enableChunking; + } + + 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 8347710a..38ca7078 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 @@ -137,6 +137,17 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ tmpProperties.setListenerName(listenerName); } + String enableChunkingStr = PropertiesUtils.getProperty(properties, PulsarMQConstants.PULSARMQ_ENABLE_CHUNKING); + if (!StringUtils.isEmpty(enableChunkingStr)) { + tmpProperties.setEnableChunking(Boolean.parseBoolean(enableChunkingStr)); + } + + String compressionType = PropertiesUtils.getProperty(properties, PulsarMQConstants.PULSARMQ_COMPRESSION_TYPE); + if (!StringUtils.isEmpty(compressionType)) { + tmpProperties.setCompressionType(compressionType); + } + + if (logger.isDebugEnabled()) { logger.debug("Load pulsar properties ==> {}", JSON.toJSON(this.mqProperties)); } @@ -408,11 +419,34 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ } // 创建指定topic的生产者 - producer = client.newProducer() - .topic(fullTopic) + ProducerBuilder producerBuilder = client.newProducer(); + if(pulsarMQProperties.getEnableChunking()){ + producerBuilder.enableChunking(true); + producerBuilder.enableBatching(false); + } + + if(!StringUtils.isEmpty(pulsarMQProperties.getCompressionType())) { + switch(pulsarMQProperties.getCompressionType().toLowerCase()) { + case "lz4": + producerBuilder.compressionType(CompressionType.LZ4); + break; + case "zlib": + producerBuilder.compressionType(CompressionType.ZLIB); + break; + case "zstd": + producerBuilder.compressionType(CompressionType.ZSTD); + break; + case "snappy": + producerBuilder.compressionType(CompressionType.SNAPPY); + break; + } + } + + producer = producerBuilder.topic(fullTopic) // 指定路由器 .messageRouter(new MessageRouterImpl(topic)) .create(); + // 放入缓存 PRODUCERS.put(topic, producer); } diff --git a/pom.xml b/pom.xml index 1940e887..6fb7bf26 100644 --- a/pom.xml +++ b/pom.xml @@ -99,13 +99,13 @@ 1.8 UTF-8 true - 5.3.9 + 5.3.26 2.17.0 - 4.8.0 + 4.9.8 5.18.0 1.0.3 2.4.0 - 2.8.1 + 2.11.4 5.1.48 0.8.3 2.22.1