chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

add compressionType,enableChunking for pulsar client , upgrade dependencies for security (#5191)

Browse Source 
* add support for listener name which is described in https://pulsar.apache.org/docs/next/concepts-multiple-advertised-listeners/

* upgrade for CVE-2023-20883  CVE-2022-1471  CVE-2023-20860

* add   enableChunking

* downgrade  pulsar client version

* fix

* 增加压缩算法
This commit is contained in:
JZ
2024-07-16 14:59:29 +08:00
committed by GitHub
parent 3439c48948
commit fb3d2c6811
7 changed files with 84 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQConstants.java
View File

@@ -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";
}

24
connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQProducerConfig.java
View File

@@ -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;
}
}

38
connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/producer/CanalPulsarMQProducer.java
View File

@@ -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);
}