From 6965726dfeaba4f175d23c2d77e11bd7553506b2 Mon Sep 17 00:00:00 2001 From: wanglei Date: Fri, 13 May 2022 12:39:43 +0800 Subject: [PATCH] fix bug #2023 (#3871) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix bug #2023 * 配置canal-adapter，docker生成脚本 --- client-adapter/docker/Dockerfile | 22 ++++++ client-adapter/docker/build.sh | 20 +++++ client-adapter/docker/startup.sh | 79 +++++++++++++++++++ .../adapter/es6x/support/ES6xTemplate.java | 8 +- .../adapter/es7x/support/ES7xTemplate.java | 11 ++- .../es/core/service/ESSyncService.java | 8 +- .../adapter/es/core/support/ESTemplate.java | 6 +- 7 files changed, 140 insertions(+), 14 deletions(-) create mode 100644 client-adapter/docker/Dockerfile create mode 100644 client-adapter/docker/build.sh create mode 100644 client-adapter/docker/startup.sh diff --git a/client-adapter/docker/Dockerfile b/client-adapter/docker/Dockerfile new file mode 100644 index 00000000..abe472c6 --- /dev/null +++ b/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 若虚 + +# 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"] \ No newline at end of file diff --git a/client-adapter/docker/build.sh b/client-adapter/docker/build.sh new file mode 100644 index 00000000..87741110 --- /dev/null +++ b/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/ diff --git a/client-adapter/docker/startup.sh b/client-adapter/docker/startup.sh new file mode 100644 index 00000000..adbc8884 --- /dev/null +++ b/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 diff --git a/client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/support/ES6xTemplate.java b/client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/support/ES6xTemplate.java index ee74f466..2e81b905 100644 --- a/client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/support/ES6xTemplate.java +++ b/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 dmlData, + public Object getESDataFromDmlData(ESSyncConfig.ESMapping mapping,String owner, Map dmlData, Map dmlOld, Map 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); diff --git a/client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/support/ES7xTemplate.java b/client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/support/ES7xTemplate.java index 8dd37761..434cbf9a 100644 --- a/client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/support/ES7xTemplate.java +++ b/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 dmlData, - Map esFieldData) { + public Object getESDataFromDmlData(ESMapping mapping, Map dmlData, Map 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 dmlData, Map dmlOld, + public Object getESDataFromDmlData(ESMapping mapping,String owner, Map dmlData, Map dmlOld, Map 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); diff --git a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/service/ESSyncService.java b/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/service/ESSyncService.java index 3ba71efd..3d75aa20 100644 --- a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/service/ESSyncService.java +++ b/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 data, + private void singleTableSimpleFiledUpdate(ESSyncConfig config,String owner, Dml dml, Map data, Map old) { ESMapping mapping = config.getEsMapping(); Map 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: {}", diff --git a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/support/ESTemplate.java b/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/support/ESTemplate.java index 2286b915..45fb67de 100644 --- a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/support/ESTemplate.java +++ b/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 dmlOld, - Map esFieldData) throws SQLException; + Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map dmlOld, Map esFieldData) throws SQLException; Object getValFromData(ESMapping mapping, Map dmlData, String fieldName, String columnName); Object getESDataFromDmlData(ESMapping mapping, Map dmlData, Map esFieldData); - Object getESDataFromDmlData(ESMapping mapping, Map dmlData, Map dmlOld, - Map esFieldData); + Object getESDataFromDmlData(ESMapping mapping,String owner, Map dmlData, Map dmlOld, Map esFieldData); }