From 8af1301062ae1b9402ed2c26279027513de2d3d7 Mon Sep 17 00:00:00 2001
From: Renkai <gaelookair@gmail.com>
Date: Tue, 13 Nov 2018 18:29:25 +0800
Subject: [PATCH] =?UTF-8?q?Kafka=E5=A2=9E=E5=8A=A0=E5=8F=82=E6=95=B0acks?=
 =?UTF-8?q?=E5=92=8CcompressionType?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../otter/canal/deployer/CanalConstants.java  |  2 ++
 .../otter/canal/deployer/CanalLauncher.java   |  2 ++
 .../otter/canal/common/MQProperties.java      | 19 +++++++++++++++++++
 .../otter/canal/kafka/CanalKafkaProducer.java |  3 ++-
 4 files changed, 25 insertions(+), 1 deletion(-)

diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java
index 2bfb32ea..aa9b2909 100644
--- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java
+++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java
@@ -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);
diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java
index 4b85f50c..f64e6184 100644
--- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java
+++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java
@@ -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;
     }
 
diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java
index 29375e2f..3bc3394b 100644
--- a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java
+++ b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java
@@ -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;
+    }
+
 }
diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
index 78a67c13..d7cf05bb 100644
--- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
+++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
@@ -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());