diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java index 1ae8c638..90e13ef8 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java +++ b/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 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 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 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 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 dmls) { if (dmls == null || dmls.isEmpty()) { diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/monitor/ESConfigMonitor.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/monitor/ESConfigMonitor.java index 23f7bfb5..cc41e728 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/monitor/ESConfigMonitor.java +++ b/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 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) {