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

Kafka增加参数acks和compressionType

Browse Source
This commit is contained in:
Renkai
2018-11-13 18:29:25 +08:00
parent 154ba422f2
commit 8af1301062
4 changed files with 25 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java
View File

@@ -45,6 +45,8 @@ public class CanalConstants {
public static final String CANAL_MQ_CANALBATCHSIZE = ROOT + "." + "mq.canalBatchSize";
public static final String CANAL_MQ_CANALGETTIMEOUT = ROOT + "." + "mq.canalGetTimeout";
public static final String CANAL_MQ_FLATMESSAGE = ROOT + "." + "mq.flatMessage";
public static final String CANAL_MQ_COMPRESSION_TYPE = ROOT + "." + "mq.compressionType";
public static final String CANAL_MQ_ACKS = ROOT + "." + "mq.acks";
public static String getInstanceModeKey(String destination) {
return MessageFormat.format(INSTANCE_MODE_TEMPLATE, destination);

2
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java
View File

@@ -100,6 +100,8 @@ public class CanalLauncher {
CanalConstants.CANAL_MQ_CANALGETTIMEOUT)));
mqProperties.setFlatMessage(Boolean.valueOf(CanalController.getProperty(properties,
CanalConstants.CANAL_MQ_FLATMESSAGE)));
mqProperties.setCompressionType(CanalController.getProperty(properties,CanalConstants.CANAL_MQ_COMPRESSION_TYPE));
mqProperties.setAcks(CanalController.getProperty(properties,CanalConstants.CANAL_MQ_ACKS));
return mqProperties;
}

19
server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java
View File

@@ -20,6 +20,8 @@ public class MQProperties {
private int canalBatchSize = 50;
private Long canalGetTimeout = 100L;
private boolean flatMessage = true;
private String compressionType = "none";
private String acks = "all";
public static class CanalDestination {
@@ -149,4 +151,21 @@ public class MQProperties {
public void setProducerGroup(String producerGroup) {
this.producerGroup = producerGroup;
}
public String getCompressionType() {
return compressionType;
}
public void setCompressionType(String compressionType) {
this.compressionType = compressionType;
}
public String getAcks() {
return acks;
}
public void setAcks(String acks) {
this.acks = acks;
}
}

3
server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
View File

@@ -37,7 +37,8 @@ public class CanalKafkaProducer implements CanalMQProducer {
this.kafkaProperties = kafkaProperties;
Properties properties = new Properties();
properties.put("bootstrap.servers", kafkaProperties.getServers());
properties.put("acks", "all");
properties.put("acks", kafkaProperties.getAcks());
properties.put("compression.type",kafkaProperties.getCompressionType());
properties.put("retries", kafkaProperties.getRetries());
properties.put("batch.size", kafkaProperties.getBatchSize());
properties.put("linger.ms", kafkaProperties.getLingerMs());