Merge pull request #1418 from rewerma/master

fixed #1416

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/AdapterRemoteConfigMonitor.java

@@ -69,6 +69,7 @@ public class AdapterRemoteConfigMonitor {
                 if (configItem.getModifiedTime() != currentConfigTimestamp) {
                     currentConfigTimestamp = configItem.getModifiedTime();
                     overrideLocalCanalConfig(configItem.getContent());
+                    logger.info("## Loaded remote adapter config: application.yml");
                 }
             }
         } catch (Exception e) {
@@ -140,7 +141,7 @@ public class AdapterRemoteConfigMonitor {
      */
     public void loadRemoteAdapterConfigs() {
         try {
-            // 加载远程instance配置
+            // 加载远程adapter配置
             Map<String, ConfigItem>[] modifiedConfigs = getModifiedAdapterConfigs();
             if (modifiedConfigs != null) {
                 overrideLocalAdapterConfigs(modifiedConfigs);
@@ -191,7 +192,7 @@ public class AdapterRemoteConfigMonitor {
                 }
             }
             if (!changedIds.isEmpty()) {
-                Map<String, ConfigItem> changedInstanceConfig = new HashMap<>();
+                Map<String, ConfigItem> changedAdapterConfig = new HashMap<>();
                 String contentsSql = "select id, category, name, content, modified_time from canal_adapter_config  where id in ("
                                      + Joiner.on(",").join(changedIds) + ")";
                 try (Statement stmt = getConn().createStatement(); ResultSet rs = stmt.executeQuery(contentsSql)) {
@@ -205,26 +206,26 @@ public class AdapterRemoteConfigMonitor {
 
                         remoteAdapterConfigs.put(configItemNew.getCategory() + "/" + configItemNew.getName(),
                             configItemNew);
-                        changedInstanceConfig.put(configItemNew.getCategory() + "/" + configItemNew.getName(),
+                        changedAdapterConfig.put(configItemNew.getCategory() + "/" + configItemNew.getName(),
                             configItemNew);
                     }
 
-                    res[0] = changedInstanceConfig.isEmpty() ? null : changedInstanceConfig;
+                    res[0] = changedAdapterConfig.isEmpty() ? null : changedAdapterConfig;
                 } catch (Exception e) {
                     logger.error(e.getMessage(), e);
                 }
             }
         }
 
-        Map<String, ConfigItem> removedInstanceConfig = new HashMap<>();
+        Map<String, ConfigItem> removedAdapterConfig = new HashMap<>();
         for (ConfigItem configItem : remoteAdapterConfigs.values()) {
             if (!remoteConfigStatus.containsKey(configItem.getCategory() + "/" + configItem.getName())) {
                 // 删除
                 remoteAdapterConfigs.remove(configItem.getCategory() + "/" + configItem.getName());
-                removedInstanceConfig.put(configItem.getCategory() + "/" + configItem.getName(), null);
+                removedAdapterConfig.put(configItem.getCategory() + "/" + configItem.getName(), null);
             }
         }
-        res[1] = removedInstanceConfig.isEmpty() ? null : removedInstanceConfig;
+        res[1] = removedAdapterConfig.isEmpty() ? null : removedAdapterConfig;
 
         if (res[0] == null && res[1] == null) {
             return null;
@@ -236,32 +237,56 @@ public class AdapterRemoteConfigMonitor {
     /**
      * 覆盖adapter配置到本地
      *
-     * @param modifiedInstanceConfigs 变动的配置集合
+     * @param modifiedAdapterConfigs 变动的配置集合
      */
-    private void overrideLocalAdapterConfigs(Map<String, ConfigItem>[] modifiedInstanceConfigs) {
+    private void overrideLocalAdapterConfigs(Map<String, ConfigItem>[] modifiedAdapterConfigs) {
-        Map<String, ConfigItem> changedInstanceConfigs = modifiedInstanceConfigs[0];
+        Map<String, ConfigItem> changedAdapterConfigs = modifiedAdapterConfigs[0];
-        if (changedInstanceConfigs != null) {
+        if (changedAdapterConfigs != null) {
-            for (ConfigItem configItem : changedInstanceConfigs.values()) {
+            for (ConfigItem configItem : changedAdapterConfigs.values()) {
                 try (FileWriter writer = new FileWriter(
                     getConfPath() + configItem.getCategory() + "/" + configItem.getName())) {
                     writer.write(configItem.getContent());
                     writer.flush();
+                    logger.info("## Loaded remote adapter config: {}/{}",
+                        configItem.getCategory(),
+                        configItem.getName());
                 } catch (Exception e) {
                     logger.error(e.getMessage(), e);
                 }
             }
         }
-        Map<String, ConfigItem> removedInstanceConfigs = modifiedInstanceConfigs[1];
+        Map<String, ConfigItem> removedAdapterConfigs = modifiedAdapterConfigs[1];
-        if (removedInstanceConfigs != null) {
+        if (removedAdapterConfigs != null) {
-            for (String name : removedInstanceConfigs.keySet()) {
+            for (String name : removedAdapterConfigs.keySet()) {
                 File file = new File(getConfPath() + name);
                 if (file.exists()) {
-                    file.delete();
+                    deleteDir(file);
+                    logger.info("## Deleted and reloaded remote adapter config: {}", name);
                 }
             }
         }
     }
 
+    private static boolean deleteDir(File dirFile) {
+        if (!dirFile.exists()) {
+            return false;
+        }
+
+        if (dirFile.isFile()) {
+            return dirFile.delete();
+        } else {
+            File[] files = dirFile.listFiles();
+            if (files == null || files.length == 0) {
+                return dirFile.delete();
+            }
+            for (File file : files) {
+                deleteDir(file);
+            }
+        }
+
+        return dirFile.delete();
+    }
+
     /**
      * 获取conf文件夹所在路径
      *

deployer/manager_ddl.sql

@@ -33,5 +33,5 @@ INSERT INTO `canal_config` VALUES ('2', 'application.yml', 'server:\n  port: 808
 INSERT INTO `canal_instance_config` VALUES ('1', 'example', '#################################################\r\n## mysql serverId , v1.0.26+ will autoGen\r\n# canal.instance.mysql.slaveId=0\r\n\r\n# enable gtid use true/false\r\ncanal.instance.gtidon=false\r\n\r\n# position info\r\ncanal.instance.master.address=127.0.0.1:3306\r\ncanal.instance.master.journal.name=\r\ncanal.instance.master.position=\r\ncanal.instance.master.timestamp=\r\ncanal.instance.master.gtid=\r\n\r\n# rds oss binlog\r\ncanal.instance.rds.accesskey=\r\ncanal.instance.rds.secretkey=\r\ncanal.instance.rds.instanceId=\r\n\r\n# table meta tsdb info\r\ncanal.instance.tsdb.enable=true\r\n#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb\r\n#canal.instance.tsdb.dbUsername=canal\r\n#canal.instance.tsdb.dbPassword=canal\r\n\r\n#canal.instance.standby.address =\r\n#canal.instance.standby.journal.name =\r\n#canal.instance.standby.position =\r\n#canal.instance.standby.timestamp =\r\n#canal.instance.standby.gtid=\r\n\r\n# username/password\r\ncanal.instance.dbUsername=canal\r\ncanal.instance.dbPassword=canal\r\ncanal.instance.connectionCharset = UTF-8\r\ncanal.instance.defaultDatabaseName =test\r\n# enable druid Decrypt database password\r\ncanal.instance.enableDruid=false\r\n#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==\r\n\r\n# table regex\r\ncanal.instance.filter.regex=.*\\\\..*\r\n# table black regex\r\ncanal.instance.filter.black.regex=\r\n\r\n# mq config\r\ncanal.mq.topic=example\r\n# 动态topic, 需mq支持动态创建topic\r\n#canal.mq.dynamicTopic=.*,mytest\\\\..*,mytest2.user\r\ncanal.mq.partition=0\r\n# hash partition config\r\n#canal.mq.partitionsNum=3\r\n#canal.mq.partitionHash=test.table:id^name,.*\\\\..*\r\n#################################################\r\n', '2018-12-30 16:46:16');
 
 INSERT INTO `canal_adapter_config` VALUES ('1', 'es', 'mytest_user.yml', 'dataSourceKey: defaultDS\ndestination: example\nesMapping:\n  _index: mytest_user\n  _type: _doc\n  _id: _id\n#  pk: id\n  sql: \"select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name,\n        a.c_time as _c_time, c.labels as _labels from user a\n        left join role b on b.id=a.role_id\n        left join (select user_id, group_concat(label order by id desc separator \';\') as labels from label\n        group by user_id) c on c.user_id=a.id\"\n#  objFields:\n#    _labels: array:;\n  etlCondition: \"where a.c_time>=\'{0}\'\"\n  commitBatch: 3000\n', '2019-01-01 17:05:43');
-INSERT INTO `canal_adapter_config` VALUES ('2', 'hhase', 'mytest_person2.yml', 'dataSourceKey: defaultDS\ndestination: example\nhbaseMapping:\n  mode: STRING  #NATIVE   #PHOENIX\n  database: mytest  # 数据库名\n  table: person2     # 数据库表名\n  hbaseTable: MYTEST.PERSON2   # HBase表名\n  family: CF  # 默认统一Family名称\n  uppercaseQualifier: true  # 字段名转大写, 默认为true\n  commitBatch: 3000 # 批量提交的大小\n  #rowKey: id,type  # 复合字段rowKey不能和columns中的rowKey重复\n  columns:\n    # 数据库字段:HBase对应字段\n    id: ROWKEY LEN:15\n    name: NAME\n    email: EMAIL\n    type:\n    c_time: C_TIME\n    birthday: BIRTHDAY\n#  excludeColumns:\n#    - lat   # 忽略字段\n\n# -- NATIVE类型\n# $DEFAULT\n# $STRING\n# $INTEGER\n# $LONG\n# $SHORT\n# $BOOLEAN\n# $FLOAT\n# $DOUBLE\n# $BIGDECIMAL\n# $DATE\n# $BYTE\n# $BYTES\n\n# -- PHOENIX类型\n# $DEFAULT                  对应PHOENIX里的VARCHAR\n# $UNSIGNED_INT             对应PHOENIX里的UNSIGNED_INT           4字节\n# $UNSIGNED_LONG            对应PHOENIX里的UNSIGNED_LONG          8字节\n# $UNSIGNED_TINYINT         对应PHOENIX里的UNSIGNED_TINYINT       1字节\n# $UNSIGNED_SMALLINT        对应PHOENIX里的UNSIGNED_SMALLINT      2字节\n# $UNSIGNED_FLOAT           对应PHOENIX里的UNSIGNED_FLOAT         4字节\n# $UNSIGNED_DOUBLE          对应PHOENIX里的UNSIGNED_DOUBLE        8字节\n# $INTEGER                  对应PHOENIX里的INTEGER                4字节\n# $BIGINT                   对应PHOENIX里的BIGINT                 8字节\n# $TINYINT                  对应PHOENIX里的TINYINT                1字节\n# $SMALLINT                 对应PHOENIX里的SMALLINT               2字节\n# $FLOAT                    对应PHOENIX里的FLOAT                  4字节\n# $DOUBLE                    对应PHOENIX里的DOUBLE                 8字节\n# $BOOLEAN                  对应PHOENIX里的BOOLEAN                1字节\n# $TIME                     对应PHOENIX里的TIME                   8字节\n# $DATE                     对应PHOENIX里的DATE                   8字节\n# $TIMESTAMP                对应PHOENIX里的TIMESTAMP              12字节\n# $UNSIGNED_TIME            对应PHOENIX里的UNSIGNED_TIME          8字节\n# $UNSIGNED_DATE            对应PHOENIX里的UNSIGNED_DATE          8字节\n# $UNSIGNED_TIMESTAMP       对应PHOENIX里的UNSIGNED_TIMESTAMP     12字节\n# $VARCHAR                  对应PHOENIX里的VARCHAR                动态长度\n# $VARBINARY                对应PHOENIX里的VARBINARY              动态长度\n# $DECIMAL                  对应PHOENIX里的DECIMAL                动态长度', '2019-01-01 17:08:08');
+INSERT INTO `canal_adapter_config` VALUES ('2', 'hbase', 'mytest_person2.yml', 'dataSourceKey: defaultDS\ndestination: example\nhbaseMapping:\n  mode: STRING  #NATIVE   #PHOENIX\n  database: mytest  # 数据库名\n  table: person2     # 数据库表名\n  hbaseTable: MYTEST.PERSON2   # HBase表名\n  family: CF  # 默认统一Family名称\n  uppercaseQualifier: true  # 字段名转大写, 默认为true\n  commitBatch: 3000 # 批量提交的大小\n  #rowKey: id,type  # 复合字段rowKey不能和columns中的rowKey重复\n  columns:\n    # 数据库字段:HBase对应字段\n    id: ROWKEY LEN:15\n    name: NAME\n    email: EMAIL\n    type:\n    c_time: C_TIME\n    birthday: BIRTHDAY\n#  excludeColumns:\n#    - lat   # 忽略字段\n\n# -- NATIVE类型\n# $DEFAULT\n# $STRING\n# $INTEGER\n# $LONG\n# $SHORT\n# $BOOLEAN\n# $FLOAT\n# $DOUBLE\n# $BIGDECIMAL\n# $DATE\n# $BYTE\n# $BYTES\n\n# -- PHOENIX类型\n# $DEFAULT                  对应PHOENIX里的VARCHAR\n# $UNSIGNED_INT             对应PHOENIX里的UNSIGNED_INT           4字节\n# $UNSIGNED_LONG            对应PHOENIX里的UNSIGNED_LONG          8字节\n# $UNSIGNED_TINYINT         对应PHOENIX里的UNSIGNED_TINYINT       1字节\n# $UNSIGNED_SMALLINT        对应PHOENIX里的UNSIGNED_SMALLINT      2字节\n# $UNSIGNED_FLOAT           对应PHOENIX里的UNSIGNED_FLOAT         4字节\n# $UNSIGNED_DOUBLE          对应PHOENIX里的UNSIGNED_DOUBLE        8字节\n# $INTEGER                  对应PHOENIX里的INTEGER                4字节\n# $BIGINT                   对应PHOENIX里的BIGINT                 8字节\n# $TINYINT                  对应PHOENIX里的TINYINT                1字节\n# $SMALLINT                 对应PHOENIX里的SMALLINT               2字节\n# $FLOAT                    对应PHOENIX里的FLOAT                  4字节\n# $DOUBLE                    对应PHOENIX里的DOUBLE                 8字节\n# $BOOLEAN                  对应PHOENIX里的BOOLEAN                1字节\n# $TIME                     对应PHOENIX里的TIME                   8字节\n# $DATE                     对应PHOENIX里的DATE                   8字节\n# $TIMESTAMP                对应PHOENIX里的TIMESTAMP              12字节\n# $UNSIGNED_TIME            对应PHOENIX里的UNSIGNED_TIME          8字节\n# $UNSIGNED_DATE            对应PHOENIX里的UNSIGNED_DATE          8字节\n# $UNSIGNED_TIMESTAMP       对应PHOENIX里的UNSIGNED_TIMESTAMP     12字节\n# $VARCHAR                  对应PHOENIX里的VARCHAR                动态长度\n# $VARBINARY                对应PHOENIX里的VARBINARY              动态长度\n# $DECIMAL                  对应PHOENIX里的DECIMAL                动态长度', '2019-01-01 17:08:08');
 INSERT INTO `canal_adapter_config` VALUES ('3', 'rdb', 'mytest_user.yml', 'dataSourceKey: defaultDS\ndestination: example\nouterAdapterKey: oracle1\nconcurrent: true\ndbMapping:\n  database: mytest\n  table: user\n  targetTable: mytest.tb_user\n  targetPk:\n    id: id\n#  mapAll: true\n  targetColumns:\n    id:\n    name:\n    role_id:\n    c_time:\n    test1:\n\n\n# Mirror schema synchronize config\n#dataSourceKey: defaultDS\n#destination: example\n#outerAdapterKey: mysql1\n#concurrent: true\n#dbMapping:\n#  mirrorDb: true\n#  database: mytest\n', '2019-01-01 17:08:50');

deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerRemoteConfigMonitor.java

@@ -4,16 +4,8 @@ import java.io.ByteArrayInputStream;
 import java.io.File;
 import java.io.FileWriter;
 import java.nio.charset.StandardCharsets;
-import java.sql.Connection;
+import java.sql.*;
-import java.sql.DriverManager;
+import java.util.*;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Statement;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -47,7 +39,7 @@ public class ManagerRemoteConfigMonitor {
 
     private long                     scanIntervalInSecond   = 5;
     private ScheduledExecutorService executor               = Executors.newScheduledThreadPool(2,
-                                                                new NamedThreadFactory("remote-canal-config-scan"));
+        new NamedThreadFactory("remote-canal-config-scan"));
 
     public ManagerRemoteConfigMonitor(String jdbcUrl, String jdbcUsername, String jdbcPassword){
         this.jdbcUrl = jdbcUrl;
@@ -83,8 +75,9 @@ public class ManagerRemoteConfigMonitor {
                     overrideLocalCanalConfig(configItem.getContent());
                     properties = new Properties();
                     properties.load(new ByteArrayInputStream(configItem.getContent().getBytes(StandardCharsets.UTF_8)));
-                    scanIntervalInSecond = Integer.valueOf(properties.getProperty(CanalConstants.CANAL_AUTO_SCAN_INTERVAL,
+                    scanIntervalInSecond = Integer
-                        "5"));
+                        .valueOf(properties.getProperty(CanalConstants.CANAL_AUTO_SCAN_INTERVAL, "5"));
+                    logger.info("## Loaded remote canal config: canal.properties ");
                 }
             }
         } catch (Exception e) {
@@ -231,9 +224,15 @@ public class ManagerRemoteConfigMonitor {
         Map<String, ConfigItem> changedInstanceConfigs = modifiedInstanceConfigs[0];
         if (changedInstanceConfigs != null) {
             for (ConfigItem configItem : changedInstanceConfigs.values()) {
-                try (FileWriter writer = new FileWriter(getConfPath() + configItem.getName() + "/instance.properties")) {
+                File instanceDir = new File(getConfPath() + configItem.getName());
+                if (!instanceDir.exists()) {
+                    instanceDir.mkdirs();
+                }
+                try (FileWriter writer = new FileWriter(
+                    getConfPath() + configItem.getName() + "/instance.properties")) {
                     writer.write(configItem.getContent());
                     writer.flush();
+                    logger.info("## Loaded remote instance config: {}/instance.properties ", configItem.getName());
                 } catch (Exception e) {
                     logger.error(e.getMessage(), e);
                 }
@@ -244,12 +243,33 @@ public class ManagerRemoteConfigMonitor {
             for (String name : removedInstanceConfigs.keySet()) {
                 File file = new File(getConfPath() + name + "/");
                 if (file.exists()) {
-                    file.delete();
+                    deleteDir(file);
+                    logger.info("## Deleted and loaded remote instance config: {} ", name);
                 }
             }
         }
     }
 
+    private static boolean deleteDir(File dirFile) {
+        if (!dirFile.exists()) {
+            return false;
+        }
+
+        if (dirFile.isFile()) {
+            return dirFile.delete();
+        } else {
+            File[] files = dirFile.listFiles();
+            if (files == null || files.length == 0) {
+                return dirFile.delete();
+            }
+            for (File file : files) {
+                deleteDir(file);
+            }
+        }
+
+        return dirFile.delete();
+    }
+
     /**
      * 监听 canal 主配置和 instance 配置变化
      *