Merge pull request #35 from alibaba/master

merge

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/db_schema.sql

@@ -7,7 +7,7 @@ CREATE TABLE `label` (
   `user_id` bigint(20) NOT NULL,
   `label` varchar(30) NOT NULL,
   PRIMARY KEY (`id`)
-) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8;
+) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4;
 
 -- ----------------------------
 -- Table structure for role
@@ -17,7 +17,7 @@ CREATE TABLE `role` (
   `id` bigint(20) NOT NULL AUTO_INCREMENT,
   `role_name` varchar(30) NOT NULL,
   PRIMARY KEY (`id`)
-) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8;
+) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4;
 
 -- ----------------------------
 -- Table structure for user
@@ -29,7 +29,7 @@ CREATE TABLE `user` (
   `c_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
   `role_id` bigint(20) DEFAULT NULL,
   PRIMARY KEY (`id`)
-) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8;
+) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4;
 
 insert into user (id,name,role_id) values (1,'Eric',1);
 insert into role (id,role_name) values (1,'admin');

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)) {

deployer/src/main/resources/canal.properties

@@ -78,8 +78,8 @@ canal.instance.tsdb.snapshot.interval = 24
 canal.instance.tsdb.snapshot.expire = 360
 
 # aliyun ak/sk , support rds/mq
-canal.aliyun.accesskey =
-canal.aliyun.secretkey =
+canal.aliyun.accessKey =
+canal.aliyun.secretKey =
 
 #################################################
 ######### 		destinations		############# 
@@ -117,4 +117,4 @@ canal.mq.compressionType = none
 canal.mq.acks = all
 # use transaction for kafka flatMessage batch produce
 canal.mq.transaction = false
-#canal.mq.properties. =
+#canal.mq.properties. =

deployer/src/main/resources/spring/tsdb/sql/create_table.sql

@@ -14,7 +14,7 @@ CREATE TABLE IF NOT EXISTS `meta_snapshot` (
   KEY `destination` (`destination`),
   KEY `destination_timestamp` (`destination`,`binlog_timestamp`),
   KEY `gmt_modified` (`gmt_modified`)
-) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='表结构记录表快照表';
+) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='表结构记录表快照表';
 
 CREATE TABLE IF NOT EXISTS `meta_history` (
   `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',
@@ -36,4 +36,4 @@ CREATE TABLE IF NOT EXISTS `meta_history` (
   KEY `destination` (`destination`),
   KEY `destination_timestamp` (`destination`,`binlog_timestamp`),
   KEY `gmt_modified` (`gmt_modified`)
-) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='表结构变化明细表';
+) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='表结构变化明细表';