From 61726c7e0594a793255d2f16499ff7dc8a2bd8da Mon Sep 17 00:00:00 2001 From: gaecfov Date: Tue, 17 Sep 2019 09:22:55 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20ES=E9=85=8D=E7=BD=AE=E6=96=87=E4=BB=B6?= =?UTF-8?q?=E7=9B=91=E6=8E=A7=E9=97=AE=E9=A2=98=20=20(#2202)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: ES监控配置文件时如果第一次因为文件错误新增失败后，再次修改不会新增的问题 ES监控配置文件添加的Config缓存没有带Instance导致匹配不到的问题 * fix：ES配置文件监控不生效 --- .../canal/client/adapter/es/ESAdapter.java | 70 ++++++++++--------- .../adapter/es/monitor/ESConfigMonitor.java | 24 +------ 2 files changed, 41 insertions(+), 53 deletions(-) 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) {