Merge branch 'master' into feature/es2

# Conflicts:
#	client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java

client-adapter/elasticsearch/pom.xml

@@ -18,12 +18,6 @@
             <version>${project.version}</version>
             <scope>provided</scope>
         </dependency>
-        <dependency>
-            <groupId>org.yaml</groupId>
-            <artifactId>snakeyaml</artifactId>
-            <version>1.19</version>
-            <scope>provided</scope>
-        </dependency>
         <dependency>
             <groupId>com.alibaba.fastsql</groupId>
             <artifactId>fastsql</artifactId>
@@ -92,4 +86,4 @@
         </plugins>
     </build>
 
-</project>
+</project>

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

@@ -1,16 +1,17 @@
 package com.alibaba.otter.canal.client.adapter.es;
 
 import java.net.InetAddress;
-import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
 import javax.sql.DataSource;
 
+import org.apache.commons.lang.StringUtils;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.client.transport.TransportClient;
 import org.elasticsearch.common.settings.Settings;
@@ -28,11 +29,7 @@ import com.alibaba.otter.canal.client.adapter.es.monitor.ESConfigMonitor;
 import com.alibaba.otter.canal.client.adapter.es.service.ESEtlService;
 import com.alibaba.otter.canal.client.adapter.es.service.ESSyncService;
 import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate;
-import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
-import com.alibaba.otter.canal.client.adapter.support.Dml;
-import com.alibaba.otter.canal.client.adapter.support.EtlResult;
-import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
-import com.alibaba.otter.canal.client.adapter.support.SPI;
+import com.alibaba.otter.canal.client.adapter.support.*;
 
 /**
  * ES外部适配器
@@ -52,6 +49,8 @@ public class ESAdapter implements OuterAdapter {
 
     private ESConfigMonitor                        esConfigMonitor;
 
+    private Properties                             envProperties;
+
     public TransportClient getTransportClient() {
         return transportClient;
     }
@@ -69,9 +68,10 @@ public class ESAdapter implements OuterAdapter {
     }
 
     @Override
-    public void init(OuterAdapterConfig configuration) {
+    public void init(OuterAdapterConfig configuration, Properties envProperties) {
         try {
-            Map<String, ESSyncConfig> esSyncConfigTmp = ESSyncConfigLoader.load();
+            this.envProperties = envProperties;
+            Map<String, ESSyncConfig> esSyncConfigTmp = ESSyncConfigLoader.load(envProperties);
             // 过滤不匹配的key的配置
             esSyncConfigTmp.forEach((key, config) -> {
                 if ((config.getOuterAdapterKey() == null && configuration.getKey() == null)
@@ -99,8 +99,21 @@ public class ESAdapter implements OuterAdapter {
                 String schema = matcher.group(2);
 
                 schemaItem.getAliasTableItems().values().forEach(tableItem -> {
-                    Map<String, ESSyncConfig> esSyncConfigMap = dbTableEsSyncConfig
-                        .computeIfAbsent(schema + "-" + tableItem.getTableName(), k -> new HashMap<>());
+                    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);
                 });
             }
@@ -138,8 +151,17 @@ public class ESAdapter implements OuterAdapter {
     public void sync(Dml dml) {
         String database = dml.getDatabase();
         String table = dml.getTable();
-        Map<String, ESSyncConfig> configMap = dbTableEsSyncConfig.get(database + "-" + table);
-        if (configMap != null) {
+        Map<String, ESSyncConfig> configMap;
+        if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
+            configMap = dbTableEsSyncConfig
+                .get(StringUtils.trimToEmpty(dml.getDestination()) + "-" + StringUtils.trimToEmpty(dml.getGroupId())
+                     + "_" + database + "-" + table);
+        } else {
+            configMap = dbTableEsSyncConfig
+                .get(StringUtils.trimToEmpty(dml.getDestination()) + "_" + database + "-" + table);
+        }
+
+        if (configMap != null && !configMap.values().isEmpty()) {
             esSyncService.sync(configMap.values(), dml);
         }
     }
@@ -208,6 +230,9 @@ public class ESAdapter implements OuterAdapter {
 
     @Override
     public void destroy() {
+        if (esConfigMonitor != null) {
+            esConfigMonitor.destroy();
+        }
         if (transportClient != null) {
             transportClient.close();
         }
@@ -215,10 +240,6 @@ public class ESAdapter implements OuterAdapter {
 
     @Override
     public String getDestination(String task) {
-        if (esConfigMonitor != null) {
-            esConfigMonitor.destroy();
-        }
-
         ESSyncConfig config = esSyncConfig.get(task);
         if (config != null) {
             return config.getDestination();

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

@@ -17,6 +17,8 @@ public class ESSyncConfig {
 
     private String    outerAdapterKey; // adapter key
 
+    private String    groupId;         // group id
+
     private String    destination;     // canal destination
 
     private ESMapping esMapping;
@@ -52,6 +54,14 @@ public class ESSyncConfig {
         this.outerAdapterKey = outerAdapterKey;
     }
 
+    public String getGroupId() {
+        return groupId;
+    }
+
+    public void setGroupId(String groupId) {
+        this.groupId = groupId;
+    }
+
     public String getDestination() {
         return destination;
     }

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

@@ -2,12 +2,12 @@ package com.alibaba.otter.canal.client.adapter.es.config;
 
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.Properties;
 
-import com.alibaba.fastjson.JSONObject;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.yaml.snakeyaml.Yaml;
 
+import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
 import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
 
 /**
@@ -20,17 +20,17 @@ public class ESSyncConfigLoader {
 
     private static Logger logger = LoggerFactory.getLogger(ESSyncConfigLoader.class);
 
-    @SuppressWarnings("unchecked")
-    public static synchronized Map<String, ESSyncConfig> load() {
+    public static synchronized Map<String, ESSyncConfig> load(Properties envProperties) {
         logger.info("## Start loading es mapping config ... ");
 
         Map<String, ESSyncConfig> esSyncConfig = new LinkedHashMap<>();
 
         Map<String, String> configContentMap = MappingConfigsLoader.loadConfigs("es");
         configContentMap.forEach((fileName, content) -> {
-            Map configMap = new Yaml().loadAs(content, Map.class); // yml自带的对象反射不是很稳定
-            JSONObject configJson = new JSONObject(configMap);
-            ESSyncConfig config = configJson.toJavaObject(ESSyncConfig.class);
+            ESSyncConfig config = YmlConfigBinder.bindYmlToObj(null, content, ESSyncConfig.class, null, envProperties);
+            if (config == null) {
+                return;
+            }
             try {
                 config.validate();
             } catch (Exception e) {

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

@@ -85,7 +85,7 @@ public class SchemaItem {
                     columnFields = new LinkedHashMap<>();
                     getSelectFields()
                         .forEach((fieldName, fieldItem) -> fieldItem.getColumnItems().forEach(columnItem -> {
-                            TableItem tableItem = getAliasTableItems().get(columnItem.getOwner());
+                            // TableItem tableItem = getAliasTableItems().get(columnItem.getOwner());
                             // if (!tableItem.isSubQuery()) {
                             List<FieldItem> fieldItems = columnFields.computeIfAbsent(
                                 columnItem.getOwner() + "." + columnItem.getColumnName(),

client-adapter/elasticsearch/src/main/resources/es/mytest_user.yml

@@ -1,15 +1,14 @@
 dataSourceKey: defaultDS
 destination: example
+groupId: g1
 esMapping:
   _index: mytest_user
   _type: _doc
   _id: _id
 #  pk: id
   sql: "select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name,
-        a.c_time as _c_time, c.labels as _labels from user a
-        left join role b on b.id=a.role_id
-        left join (select user_id, group_concat(label order by id desc separator ';') as labels from label
-        group by user_id) c on c.user_id=a.id"
+        a.c_time as _c_time from user a
+        left join role b on b.id=a.role_id"
 #  objFields:
 #    _labels: array:;
   etlCondition: "where a.c_time>='{0}'"

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ConfigLoadTest.java

@@ -21,7 +21,7 @@ public class ConfigLoadTest {
 
     @Test
     public void testLoad() {
-        Map<String, ESSyncConfig> configMap = ESSyncConfigLoader.load();
+        Map<String, ESSyncConfig> configMap = ESSyncConfigLoader.load(null);
         ESSyncConfig config = configMap.get("mytest_user.yml");
         Assert.assertNotNull(config);
         Assert.assertEquals("defaultDS", config.getDataSourceKey());

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/Common.java

@@ -26,7 +26,7 @@ public class Common {
         outerAdapterConfig.setProperties(properties);
 
         ESAdapter esAdapter = new ESAdapter();
-        esAdapter.init(outerAdapterConfig);
+        esAdapter.init(outerAdapterConfig, null);
         return esAdapter;
     }