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

fixed pull #5191 , reformat

Browse Source
This commit is contained in:
jianghang.loujh
2024-07-16 15:02:10 +08:00
parent fb3d2c6811
commit fea4d8210a
3 changed files with 13 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -66,7 +66,6 @@ public class PulsarMQConstants {
*/ */
public static final String PULSARMQ_LISTENER_NAME = ROOT + "." + "listenerName"; public static final String PULSARMQ_LISTENER_NAME = ROOT + "." + "listenerName";
/** /**
* Pulsar 开启chunking * Pulsar 开启chunking
*/ */

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

@@ -95,6 +95,7 @@ public class PulsarMQProducerConfig extends MQProperties {
public void setEnableChunking(boolean enableChunking) { public void setEnableChunking(boolean enableChunking) {
this.enableChunking = enableChunking; this.enableChunking = enableChunking;
} }
public boolean getEnableChunking() { public boolean getEnableChunking() {
return this.enableChunking; return this.enableChunking;
} }
@@ -102,6 +103,7 @@ public class PulsarMQProducerConfig extends MQProperties {
public void setCompressionType(String compressionType) { public void setCompressionType(String compressionType) {
this.compressionType = compressionType; this.compressionType = compressionType;
} }
public String getCompressionType() { public String getCompressionType() {
return this.compressionType; return this.compressionType;
} }

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

@@ -147,7 +147,6 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
tmpProperties.setCompressionType(compressionType); tmpProperties.setCompressionType(compressionType);
} }
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("Load pulsar properties ==> {}", JSON.toJSON(this.mqProperties)); logger.debug("Load pulsar properties ==> {}", JSON.toJSON(this.mqProperties));
} }
@@ -420,13 +419,13 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
// 创建指定topic的生产者 // 创建指定topic的生产者
ProducerBuilder producerBuilder = client.newProducer(); ProducerBuilder producerBuilder = client.newProducer();
if(pulsarMQProperties.getEnableChunking()){ if (pulsarMQProperties.getEnableChunking()) {
producerBuilder.enableChunking(true); producerBuilder.enableChunking(true);
producerBuilder.enableBatching(false); producerBuilder.enableBatching(false);
} }
if(!StringUtils.isEmpty(pulsarMQProperties.getCompressionType())) { if (!StringUtils.isEmpty(pulsarMQProperties.getCompressionType())) {
switch(pulsarMQProperties.getCompressionType().toLowerCase()) { switch (pulsarMQProperties.getCompressionType().toLowerCase()) {
case "lz4": case "lz4":
producerBuilder.compressionType(CompressionType.LZ4); producerBuilder.compressionType(CompressionType.LZ4);
break; break;