Merge pull request #1567 from dragontree101/miss_producergroup_properties

producerGroup配置没有读取

deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java

@@ -49,6 +49,7 @@ public class CanalConstants {
     public static final String CANAL_MQ_COMPRESSION_TYPE         = ROOT + "." + "mq.compressionType";
     public static final String CANAL_MQ_ACKS                     = ROOT + "." + "mq.acks";
     public static final String CANAL_MQ_TRANSACTION              = ROOT + "." + "mq.transaction";
+    public static final String CANAL_MQ_PRODUCERGROUP            = ROOT + "." + "mq.producerGroup";
     public static final String CANAL_ALIYUN_ACCESSKEY            = ROOT + "." + "aliyun.accessKey";
     public static final String CANAL_ALIYUN_SECRETKEY            = ROOT + "." + "aliyun.secretKey";
     public static final String CANAL_MQ_PROPERTIES               = ROOT + "." + "mq.properties";

deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java

@@ -164,6 +164,11 @@ public class CanalStater {
             mqProperties.setTransaction(Boolean.valueOf(transaction));
         }
 
+        String producerGroup = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_PRODUCERGROUP);
+        if (!StringUtils.isEmpty(producerGroup)) {
+            mqProperties.setProducerGroup(producerGroup);
+        }
+
         for (Object key : properties.keySet()) {
             key = StringUtils.trim(key.toString());
             if (((String) key).startsWith(CanalConstants.CANAL_MQ_PROPERTIES)) {