chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat 元数据与target数据源不一致时 进行兼容处理 (#5168)

Browse Source 
* feat 元数据与target数据源不一致时 进行兼容处理

* feat 元数据与target数据源不一致时 进行兼容处理  优化代码

* Feature #FR-761 ddl执行失败会造成后续未执行的dml丢弃 针对ddl exec failed catch the ex
This commit is contained in:
aiden.liu
2024-06-26 15:43:10 +08:00
committed by GitHub
parent 4c906702a6
commit c1f5a5a8e4
4 changed files with 78 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java
View File

@@ -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());
}
}
}

96
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java
View File

@@ -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<String, Integer> ctype = getTargetColumnType(batchExecutor.getConn(), config);
Map<String, Integer> ctype = getTargetColumnTypeByColumns(batchExecutor.getConn(), config, columnsMap);
List<Map<String, ?>> values = new ArrayList<>();
for (Map.Entry<String, String> entry : columnsMap.entrySet()) {
@@ -328,7 +329,7 @@ public class RdbSyncService {
String backtick = SyncUtil.getBacktickByDbType(dataSource.getDbType());
Map<String, String> columnsMap = SyncUtil.getColumnsMap(dbMapping, data);
Map<String, Integer> ctype = getTargetColumnType(batchExecutor.getConn(), config);
Map<String, Integer> 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<String, Integer> getTargetColumnTypeByColumns(Connection conn, MappingConfig config,
Map<String, String> columnsMap) {
DbMapping dbMapping = config.getDbMapping();
String cacheKey = config.getDestination() + "." + dbMapping.getDatabase() + "." + dbMapping.getTable();
Map<String, Integer> 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 <T> 集合元素类型
* @param column
* @param anotherColumn
* @return {@link Boolean}
*/
public static<T> Boolean checkCloumnConsistent(Collection<T> column, Collection<T> anotherColumn) {
if(null == column || null == anotherColumn || column.size() != anotherColumn.size()){
return false;
}
Map<T, T> anotherColumnMap = column.stream().collect(Collectors.toMap(Function.identity(), Function.identity()));
for (T t : column) {
if(null == anotherColumnMap.get(t)){
return false;
}
}
return true;
}
/**
* 获取目标字段类型
*

2
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java
View File

@@ -26,7 +26,7 @@ public class SyncUtil {
public static Map<String, String> getColumnsMap(MappingConfig.DbMapping dbMapping, Collection<String> columns) {
Map<String, String> columnsMap;
if (dbMapping.getMapAll()) {
if (dbMapping.getAllMapColumns() != null) {
if (dbMapping.getAllMapColumns() != null && columns.size() == dbMapping.getAllMapColumns().size() ) {
return dbMapping.getAllMapColumns();
}
columnsMap = new LinkedHashMap<>();

2
client-adapter/tablestore/pom.xml
View File

@@ -10,7 +10,7 @@
<groupId>com.alibaba.otter</groupId>
<artifactId>client-adapter.tablestore</artifactId>
<packaging>jar</packaging>
<name>canal client adapter rdb module for otter ${project.version}</name>
<name>canal client adapter tablestore module for otter ${project.version}</name>
<dependencies>
<dependency>