fix: ES配置文件监控问题 (#2202)

* fix:
ES监控配置文件时如果第一次因为文件错误新增失败后，再次修改不会新增的问题
ES监控配置文件添加的Config缓存没有带Instance导致匹配不到的问题

* fix：ES配置文件监控不生效

client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java

@@ -84,38 +84,7 @@ public class ESAdapter implements OuterAdapter {
             for (Map.Entry<String, ESSyncConfig> entry : esSyncConfig.entrySet()) {
                 String configName = entry.getKey();
                 ESSyncConfig config = entry.getValue();
-                SchemaItem schemaItem = SqlParser.parse(config.getEsMapping().getSql());
-                config.getEsMapping().setSchemaItem(schemaItem);
-
-                DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
-                if (dataSource == null || dataSource.getUrl() == null) {
-                    throw new RuntimeException("No data source found: " + config.getDataSourceKey());
-                }
-                Pattern pattern = Pattern.compile(".*:(.*)://.*/(.*)\\?.*$");
-                Matcher matcher = pattern.matcher(dataSource.getUrl());
-                if (!matcher.find()) {
-                    throw new RuntimeException("Not found the schema of jdbc-url: " + config.getDataSourceKey());
-                }
-                String schema = matcher.group(2);
-
-                schemaItem.getAliasTableItems().values().forEach(tableItem -> {
-                    Map<String, ESSyncConfig> esSyncConfigMap;
-                    if (envProperties != null
-                        && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
-                        esSyncConfigMap = dbTableEsSyncConfig
-                            .computeIfAbsent(StringUtils.trimToEmpty(config.getDestination()) + "-"
-                                             + StringUtils.trimToEmpty(config.getGroupId()) + "_" + schema + "-"
-                                             + tableItem.getTableName(),
-                                k -> new ConcurrentHashMap<>());
-                    } else {
-                        esSyncConfigMap = dbTableEsSyncConfig
-                            .computeIfAbsent(StringUtils.trimToEmpty(config.getDestination()) + "_" + schema + "-"
-                                             + tableItem.getTableName(),
-                                k -> new ConcurrentHashMap<>());
-                    }
-
-                    esSyncConfigMap.put(configName, config);
-                });
+                addSyncConfigToCache(configName, config);
             }
 
             Map<String, String> properties = configuration.getProperties();
@@ -138,6 +107,43 @@ public class ESAdapter implements OuterAdapter {
         }
     }
 
+    public void addSyncConfigToCache(String configName, ESSyncConfig config) {
+        Properties envProperties = this.envProperties;
+        SchemaItem schemaItem = SqlParser.parse(config.getEsMapping().getSql());
+        config.getEsMapping().setSchemaItem(schemaItem);
+
+        DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
+        if (dataSource == null || dataSource.getUrl() == null) {
+            throw new RuntimeException("No data source found: " + config.getDataSourceKey());
+        }
+        Pattern pattern = Pattern.compile(".*:(.*)://.*/(.*)\\?.*$");
+        Matcher matcher = pattern.matcher(dataSource.getUrl());
+        if (!matcher.find()) {
+            throw new RuntimeException("Not found the schema of jdbc-url: " + config.getDataSourceKey());
+        }
+        String schema = matcher.group(2);
+
+        schemaItem.getAliasTableItems().values().forEach(tableItem -> {
+            Map<String, ESSyncConfig> esSyncConfigMap;
+            if (envProperties != null
+                && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
+                esSyncConfigMap = dbTableEsSyncConfig
+                    .computeIfAbsent(StringUtils.trimToEmpty(config.getDestination()) + "-"
+                                     + StringUtils.trimToEmpty(config.getGroupId()) + "_" + schema + "-"
+                                     + tableItem.getTableName(),
+                        k -> new ConcurrentHashMap<>());
+            } else {
+                esSyncConfigMap = dbTableEsSyncConfig
+                    .computeIfAbsent(StringUtils.trimToEmpty(config.getDestination()) + "_" + schema + "-"
+                                     + tableItem.getTableName(),
+                        k -> new ConcurrentHashMap<>());
+            }
+
+            esSyncConfigMap.put(configName, config);
+        });
+    }
+
+
     @Override
     public void sync(List<Dml> dmls) {
         if (dmls == null || dmls.isEmpty()) {

client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/monitor/ESConfigMonitor.java

@@ -106,6 +106,8 @@ public class ESConfigMonitor {
                     addConfigToCache(file, config);
 
                     logger.info("Change a es mapping config: {} of canal adapter", file.getName());
+                }else{
+                    onFileCreate(file);
                 }
             } catch (Exception e) {
                 logger.error(e.getMessage(), e);
@@ -129,27 +131,7 @@ public class ESConfigMonitor {
 
         private void addConfigToCache(File file, ESSyncConfig config) {
             esAdapter.getEsSyncConfig().put(file.getName(), config);
-
-            SchemaItem schemaItem = SqlParser.parse(config.getEsMapping().getSql());
-            config.getEsMapping().setSchemaItem(schemaItem);
-
-            DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
-            if (dataSource == null || dataSource.getUrl() == null) {
-                throw new RuntimeException("No data source found: " + config.getDataSourceKey());
-            }
-            Pattern pattern = Pattern.compile(".*:(.*)://.*/(.*)\\?.*$");
-            Matcher matcher = pattern.matcher(dataSource.getUrl());
-            if (!matcher.find()) {
-                throw new RuntimeException("Not found the schema of jdbc-url: " + config.getDataSourceKey());
-            }
-            String schema = matcher.group(2);
-
-            schemaItem.getAliasTableItems().values().forEach(tableItem -> {
-                Map<String, ESSyncConfig> esSyncConfigMap = esAdapter.getDbTableEsSyncConfig()
-                    .computeIfAbsent(schema + "-" + tableItem.getTableName(), k -> new HashMap<>());
-                esSyncConfigMap.put(file.getName(), config);
-            });
-
+            esAdapter.addSyncConfigToCache(file.getName(),config);
         }
 
         private void deleteConfigFromCache(File file) {