diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/db_schema.sql b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/db_schema.sql index bff2e118..c7274ca2 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/db_schema.sql +++ b/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'); 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 f387e1b4..45cffa5a 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 @@ -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"; diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java index 05c8c6f0..a3d6035f 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java +++ b/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)) { diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 8b5c09fc..30049c00 100644 --- a/deployer/src/main/resources/canal.properties +++ b/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. = \ No newline at end of file +#canal.mq.properties. = diff --git a/deployer/src/main/resources/spring/tsdb/sql/create_table.sql b/deployer/src/main/resources/spring/tsdb/sql/create_table.sql index 322bc2b6..8cb82c82 100644 --- a/deployer/src/main/resources/spring/tsdb/sql/create_table.sql +++ b/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='表结构变化明细表'; \ No newline at end of file +) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='表结构变化明细表'; \ No newline at end of file