fix bug #2023 (#3871)

* fix bug #2023

* 配置canal-adapter，docker生成脚本

client-adapter/docker/Dockerfile

@@ -0,0 +1,22 @@
+# 本脚本参考slpcat的docker镜像生成脚本
+# 链接：https://github.com/slpcat/docker-images/tree/master/bigdata/canal/debian10-canal-adapter-jdk8
+FROM slpcat/openjdk:8
+
+MAINTAINER 若虚 <slpcat@qq.com>
+
+# install canal-adapter
+COPY canal.adapter-*.tar.gz /tmp/
+
+RUN \
+    mkdir -p /opt/canal-adapter && \
+    tar -zxf /tmp/canal.adapter-*.tar.gz -C /opt/canal-adapter && \
+    rm -r /tmp/canal.adapter-*.tar.gz
+
+COPY startup.sh /opt/canal-adapter/bin/startup.sh
+
+# 8081 exporter
+EXPOSE 8081
+
+WORKDIR /opt/canal-adapter
+
+CMD ["sh", "-c", "./bin/startup.sh"]

client-adapter/docker/build.sh

@@ -0,0 +1,20 @@
+#!/bin/bash
+
+current_path=`pwd`
+case "`uname`" in
+    Darwin)
+        bin_abs_path=`cd $(dirname $0); pwd`
+        ;;
+    Linux)
+        bin_abs_path=$(readlink -f $(dirname $0))
+        ;;
+    *)
+        bin_abs_path=`cd $(dirname $0); pwd`
+        ;;
+esac
+BASE=${bin_abs_path}
+
+rm -rf $BASE/canal.*.tar.gz ;
+cd $BASE/../../ && mvn clean package -DskipTests -Denv=release && cd $current_path ;
+cp $BASE/../../target/canal.adapter-*.tar.gz $BASE/
+docker build --no-cache -t canal/canal-adapter:v1.1.6 $BASE/

client-adapter/docker/startup.sh

@@ -0,0 +1,79 @@
+#!/bin/bash
+
+current_path=`pwd`
+case "`uname`" in
+    Linux)
+		bin_abs_path=$(readlink -f $(dirname $0))
+		;;
+	*)
+		bin_abs_path=`cd $(dirname $0); pwd`
+		;;
+esac
+base=${bin_abs_path}/..
+export LANG=en_US.UTF-8
+export BASE=$base
+
+if [ -f $base/bin/adapter.pid ] ; then
+	echo "found adapter.pid , Please run stop.sh first ,then startup.sh" 2>&2
+    exit 1
+fi
+
+if [ ! -d $base/logs ] ; then
+	mkdir -p $base/logs
+fi
+
+## set java path
+if [ -z "$JAVA" ] ; then
+  JAVA=$(which java)
+fi
+
+ALIBABA_JAVA="/usr/alibaba/java/bin/java"
+TAOBAO_JAVA="/opt/taobao/java/bin/java"
+if [ -z "$JAVA" ]; then
+  if [ -f $ALIBABA_JAVA ] ; then
+  	JAVA=$ALIBABA_JAVA
+  elif [ -f $TAOBAO_JAVA ] ; then
+  	JAVA=$TAOBAO_JAVA
+  else
+  	echo "Cannot find a Java JDK. Please set either set JAVA or put java (>=1.5) in your PATH." 2>&2
+    exit 1
+  fi
+fi
+
+case "$#"
+in
+0 )
+  ;;
+2 )
+  if [ "$1" = "debug" ]; then
+    DEBUG_PORT=$2
+    DEBUG_SUSPEND="n"
+    JAVA_DEBUG_OPT="-Xdebug -Xnoagent -Djava.compiler=NONE -Xrunjdwp:transport=dt_socket,address=$DEBUG_PORT,server=y,suspend=$DEBUG_SUSPEND"
+  fi
+  ;;
+* )
+  echo "THE PARAMETERS MUST BE TWO OR LESS.PLEASE CHECK AGAIN."
+  exit;;
+esac
+
+str=`file -L $JAVA | grep 64-bit`
+if [ -n "$str" ]; then
+	JAVA_OPTS="-server -Xms2048m -Xmx3072m -Xmn1024m -XX:SurvivorRatio=2 -Xss256k -XX:+DisableExplicitGC -XX:+HeapDumpOnOutOfMemoryError"
+else
+	JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m "
+fi
+
+JAVA_OPTS=" $JAVA_OPTS -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dfile.encoding=UTF-8"
+ADAPTER_OPTS="-DappName=canal-adapter"
+
+for i in $base/lib/*;
+    do CLASSPATH=$i:"$CLASSPATH";
+done
+
+CLASSPATH="$base/conf:$CLASSPATH";
+
+echo "cd to $bin_abs_path for workaround relative path"
+cd $bin_abs_path
+
+echo CLASSPATH :$CLASSPATH
+exec $JAVA $JAVA_OPTS $JAVA_DEBUG_OPT $ADAPTER_OPTS -classpath .:$CLASSPATH com.alibaba.otter.canal.adapter.launcher.CanalAdapterApplication

client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/support/ES6xTemplate.java

@@ -316,13 +316,17 @@ public class ES6xTemplate implements ESTemplate {
     }
 
     @Override
-    public Object getESDataFromDmlData(ESSyncConfig.ESMapping mapping, Map<String, Object> dmlData,
+    public Object getESDataFromDmlData(ESSyncConfig.ESMapping mapping,String owner, Map<String, Object> dmlData,
                                        Map<String, Object> dmlOld, Map<String, Object> esFieldData) {
         SchemaItem schemaItem = mapping.getSchemaItem();
         String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
-            String columnName = fieldItem.getColumnItems().iterator().next().getColumnName();
+            ColumnItem columnItem = fieldItem.getColumnItems().iterator().next();
+            if (!columnItem.getOwner().equals(owner)) {
+                continue;
+            }
+            String columnName = columnItem.getColumnName();
 
             if (fieldItem.getFieldName().equals(idFieldName)) {
                 resultIdVal = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName);

client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/support/ES7xTemplate.java

@@ -287,8 +287,7 @@ public class ES7xTemplate implements ESTemplate {
     }
 
     @Override
-    public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData,
-                                       Map<String, Object> esFieldData) {
+    public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData, Map<String, Object> esFieldData) {
         SchemaItem schemaItem = mapping.getSchemaItem();
         String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
         Object resultIdVal = null;
@@ -312,13 +311,17 @@ public class ES7xTemplate implements ESTemplate {
     }
 
     @Override
-    public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData, Map<String, Object> dmlOld,
+    public Object getESDataFromDmlData(ESMapping mapping,String owner, Map<String, Object> dmlData, Map<String, Object> dmlOld,
                                        Map<String, Object> esFieldData) {
         SchemaItem schemaItem = mapping.getSchemaItem();
         String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
-            String columnName = fieldItem.getColumnItems().iterator().next().getColumnName();
+            ColumnItem columnItem = fieldItem.getColumnItems().iterator().next();
+            if (!columnItem.getOwner().equals(owner)) {
+                continue;
+            }
+            String columnName = columnItem.getColumnName();
 
             if (fieldItem.getFieldName().equals(idFieldName)) {
                 resultIdVal = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName);

client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/service/ESSyncService.java

@@ -209,7 +209,7 @@ public class ESSyncService {
 
             if (schemaItem.getAliasTableItems().size() == 1 && schemaItem.isAllFieldsSimple()) {
                 // ------单表 & 所有字段都为简单字段------
-                singleTableSimpleFiledUpdate(config, dml, data, old);
+                singleTableSimpleFiledUpdate(config,schemaItem.getMainTable().getAlias(), dml, data, old);
             } else {
                 // ------主表 查询sql来更新------
                 if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) {
@@ -263,7 +263,7 @@ public class ESSyncService {
 
                     // 判断主键和所更新的字段是否全为简单字段
                     if (idFieldSimple && allUpdateFieldSimple && !fkChanged) {
-                        singleTableSimpleFiledUpdate(config, dml, data, old);
+                        singleTableSimpleFiledUpdate(config,schemaItem.getMainTable().getAlias(), dml, data, old);
                     } else {
                         mainTableUpdate(config, dml, data, old);
                     }
@@ -810,12 +810,12 @@ public class ESSyncService {
      * @param data 单行data数据
      * @param old 单行old数据
      */
-    private void singleTableSimpleFiledUpdate(ESSyncConfig config, Dml dml, Map<String, Object> data,
+    private void singleTableSimpleFiledUpdate(ESSyncConfig config,String owner, Dml dml, Map<String, Object> data,
                                               Map<String, Object> old) {
         ESMapping mapping = config.getEsMapping();
         Map<String, Object> esFieldData = new LinkedHashMap<>();
 
-        Object idVal = esTemplate.getESDataFromDmlData(mapping, data, old, esFieldData);
+        Object idVal = esTemplate.getESDataFromDmlData(mapping,owner, data, old, esFieldData);
 
         if (logger.isTraceEnabled()) {
             logger.trace("Main table update to es index, destination:{}, table: {}, index: {}, id: {}",

client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/support/ESTemplate.java

@@ -57,13 +57,11 @@ public interface ESTemplate {
 
     Object getIdValFromRS(ESMapping mapping, ResultSet resultSet) throws SQLException;
 
-    Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map<String, Object> dmlOld,
-                           Map<String, Object> esFieldData) throws SQLException;
+    Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map<String, Object> dmlOld, Map<String, Object> esFieldData) throws SQLException;
 
     Object getValFromData(ESMapping mapping, Map<String, Object> dmlData, String fieldName, String columnName);
 
     Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData, Map<String, Object> esFieldData);
 
-    Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData, Map<String, Object> dmlOld,
-                                Map<String, Object> esFieldData);
+    Object getESDataFromDmlData(ESMapping mapping,String owner, Map<String, Object> dmlData, Map<String, Object> dmlOld, Map<String, Object> esFieldData);
 }