From c1f5a5a8e45ef216c47693da1ff5a5d7d73a4092 Mon Sep 17 00:00:00 2001 From: "aiden.liu" <17628390413@163.com> Date: Wed, 26 Jun 2024 15:43:10 +0800 Subject: [PATCH] =?UTF-8?q?feat=20=E5=85=83=E6=95=B0=E6=8D=AE=E4=B8=8Etarg?= =?UTF-8?q?et=E6=95=B0=E6=8D=AE=E6=BA=90=E4=B8=8D=E4=B8=80=E8=87=B4?= =?UTF-8?q?=E6=97=B6=20=E8=BF=9B=E8=A1=8C=E5=85=BC=E5=AE=B9=E5=A4=84?= =?UTF-8?q?=E7=90=86=20(#5168)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat 元数据与target数据源不一致时 进行兼容处理 * feat 元数据与target数据源不一致时 进行兼容处理 优化代码 * Feature #FR-761 ddl执行失败会造成后续未执行的dml丢弃 针对ddl exec failed catch the ex --- .../rdb/service/RdbMirrorDbSyncService.java | 3 +- .../adapter/rdb/service/RdbSyncService.java | 96 ++++++++++++++----- .../client/adapter/rdb/support/SyncUtil.java | 2 +- client-adapter/tablestore/pom.xml | 2 +- 4 files changed, 78 insertions(+), 25 deletions(-) diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java index 54498a7e..4207b5f1 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java @@ -164,7 +164,8 @@ public class RdbMirrorDbSyncService { logger.trace("Execute DDL sql: {} for database: {}", ddl.getSql(), ddl.getDatabase()); } } catch (Exception e) { - throw new RuntimeException(e); + logger.error("Execute DDL sql failed sql: {} for database: {} cause: {}", + ddl.getSql(), ddl.getDatabase(), e.getMessage()); } } } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java index 17830e44..50a17077 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java @@ -1,25 +1,5 @@ package com.alibaba.otter.canal.client.adapter.rdb.service; -import java.sql.Connection; -import java.sql.ResultSetMetaData; -import java.sql.SQLException; -import java.sql.Types; -import java.util.ArrayList; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.Properties; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; -import java.util.function.Function; - -import org.apache.commons.lang.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONWriter.Feature; @@ -30,6 +10,27 @@ import com.alibaba.otter.canal.client.adapter.rdb.support.SingleDml; import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil; import com.alibaba.otter.canal.client.adapter.support.Dml; import com.alibaba.otter.canal.client.adapter.support.Util; +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.sql.Connection; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.sql.Types; +import java.util.ArrayList; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.function.Function; +import java.util.stream.Collectors; /** * RDB同步操作业务 @@ -272,7 +273,7 @@ public class RdbSyncService { len = insertSql.length(); insertSql.delete(len - 1, len).append(")"); - Map ctype = getTargetColumnType(batchExecutor.getConn(), config); + Map ctype = getTargetColumnTypeByColumns(batchExecutor.getConn(), config, columnsMap); List> values = new ArrayList<>(); for (Map.Entry entry : columnsMap.entrySet()) { @@ -328,7 +329,7 @@ public class RdbSyncService { String backtick = SyncUtil.getBacktickByDbType(dataSource.getDbType()); Map columnsMap = SyncUtil.getColumnsMap(dbMapping, data); - Map ctype = getTargetColumnType(batchExecutor.getConn(), config); + Map ctype = getTargetColumnTypeByColumns(batchExecutor.getConn(), config, columnsMap); StringBuilder updateSql = new StringBuilder(); updateSql.append("UPDATE ").append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType())).append(" SET "); @@ -410,6 +411,57 @@ public class RdbSyncService { } } + /** + * Obtain metadata for the corresponding table columns of the target data source + * Check if the metadata of the columns in the src is consistent with the metadata cache of the columns in the cache + * (first determine the number of columns and then determine if each column is consistent). + * If not, reload + * @param conn + * @param config + * @param columnsMap + * @date 2024/6/17 14:45 + */ + private Map getTargetColumnTypeByColumns(Connection conn, MappingConfig config, + Map columnsMap) { + DbMapping dbMapping = config.getDbMapping(); + String cacheKey = config.getDestination() + "." + dbMapping.getDatabase() + "." + dbMapping.getTable(); + Map columnType = getTargetColumnType(conn, config); + if(null != columnType && null != columnsMap && + !checkCloumnConsistent(columnType.keySet(), columnsMap.keySet())){ + logger.info("column meta data not match, reload the column meta data from target db. key:{}", cacheKey); + columnsTypeCache.remove(cacheKey); + columnType = getTargetColumnType(conn, config); + if(null != columnType && null != columnsMap && + !checkCloumnConsistent(columnType.keySet(), columnsMap.keySet())) { + logger.info("the column meta data from db still not match. dbdata:{}", columnType.keySet()); + } + } + return columnType; + } + + + /** + * 判断binlog的中的列的数据和缓存的列数据是否一致 + * + * @param 集合元素类型 + * @param column + * @param anotherColumn + * @return {@link Boolean} + */ + public static Boolean checkCloumnConsistent(Collection column, Collection anotherColumn) { + if(null == column || null == anotherColumn || column.size() != anotherColumn.size()){ + return false; + } + Map anotherColumnMap = column.stream().collect(Collectors.toMap(Function.identity(), Function.identity())); + + for (T t : column) { + if(null == anotherColumnMap.get(t)){ + return false; + } + } + return true; + } + /** * 获取目标字段类型 * diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java index c5fb4d5b..7ebeb271 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java @@ -26,7 +26,7 @@ public class SyncUtil { public static Map getColumnsMap(MappingConfig.DbMapping dbMapping, Collection columns) { Map columnsMap; if (dbMapping.getMapAll()) { - if (dbMapping.getAllMapColumns() != null) { + if (dbMapping.getAllMapColumns() != null && columns.size() == dbMapping.getAllMapColumns().size() ) { return dbMapping.getAllMapColumns(); } columnsMap = new LinkedHashMap<>(); diff --git a/client-adapter/tablestore/pom.xml b/client-adapter/tablestore/pom.xml index 0b5056be..5ee7f5d6 100644 --- a/client-adapter/tablestore/pom.xml +++ b/client-adapter/tablestore/pom.xml @@ -10,7 +10,7 @@ com.alibaba.otter client-adapter.tablestore jar - canal client adapter rdb module for otter ${project.version} + canal client adapter tablestore module for otter ${project.version}