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

Revert "修复测试类报错 (#4516)"

Browse Source 
This reverts commit d899981354.
This commit is contained in:
rewerma
2022-12-06 16:41:17 +08:00
committed by GitHub
parent d899981354
commit f3bb0dd18d
89 changed files with 4470 additions and 1105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java
View File

@@ -1,6 +1,5 @@
package com.alibaba.otter.canal.client.adapter.rdb.config;
import com.alibaba.otter.canal.client.adapter.support.YamlUtils;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Properties;
@@ -8,6 +7,7 @@ import java.util.Properties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
/**
@@ -32,7 +32,8 @@ public class ConfigLoader {
Map<String, String> configContentMap = MappingConfigsLoader.loadConfigs("rdb");
configContentMap.forEach((fileName, content) -> {
MappingConfig config = YamlUtils.ymlToObj(null, content, MappingConfig.class, null, envProperties);
MappingConfig config = YmlConfigBinder
.bindYmlToObj(null, content, MappingConfig.class, null, envProperties);
if (config == null) {
return;
}

19
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java
View File

@@ -1,10 +1,10 @@
package com.alibaba.otter.canal.client.adapter.rdb.monitor;
import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter;
import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
import com.alibaba.otter.canal.client.adapter.support.Util;
import com.alibaba.otter.canal.client.adapter.support.YamlUtils;
import java.io.File;
import java.util.Properties;
import org.apache.commons.io.filefilter.FileFilterUtils;
@@ -35,7 +35,7 @@ public class RdbConfigMonitor {
File confDir = Util.getConfDirPath(adapterName);
try {
FileAlterationObserver observer = new FileAlterationObserver(confDir,
FileFilterUtils.and(FileFilterUtils.fileFileFilter(), FileFilterUtils.suffixFileFilter("yml")));
FileFilterUtils.and(FileFilterUtils.fileFileFilter(), FileFilterUtils.suffixFileFilter("yml")));
FileListener listener = new FileListener();
observer.addListener(listener);
fileMonitor = new FileAlterationMonitor(3000, observer);
@@ -62,15 +62,16 @@ public class RdbConfigMonitor {
try {
// 加载新增的配置文件
String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName());
MappingConfig config = YamlUtils
.ymlToObj(null, configContent, MappingConfig.class, null, envProperties);
MappingConfig config = YmlConfigBinder
.bindYmlToObj(null, configContent, MappingConfig.class, null, envProperties);
if (config == null) {
return;
}
config.validate();
boolean result = rdbAdapter.addConfig(file.getName(), config);
if (result) {
logger.info("Add a new rdb mapping config: {} to canal adapter", file.getName());
logger.info("Add a new rdb mapping config: {} to canal adapter",
file.getName());
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
@@ -84,14 +85,14 @@ public class RdbConfigMonitor {
try {
if (rdbAdapter.getRdbMapping().containsKey(file.getName())) {
// 加载配置文件
String configContent = MappingConfigsLoader
.loadConfig(adapterName + File.separator + file.getName());
String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName());
if (configContent == null) {
onFileDelete(file);
return;
}
MappingConfig config = YamlUtils
.ymlToObj(null, configContent, MappingConfig.class, null, envProperties);
MappingConfig config = YmlConfigBinder
.bindYmlToObj(null, configContent, MappingConfig.class, null,
envProperties);
if (config == null) {
return;
}

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

@@ -58,7 +58,6 @@ public class RdbEtlService extends AbstractEtlService {
Map<String, String> columnsMap = new LinkedHashMap<>();
Map<String, Integer> columnType = new LinkedHashMap<>();
DruidDataSource dataSource = (DruidDataSource) srcDS;
String backtick = SyncUtil.getBacktickByDbType(dataSource.getDbType());
Util.sqlRS(targetDS,
"SELECT * FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()) + " LIMIT 1 ",
@@ -92,7 +91,7 @@ public class RdbEtlService extends AbstractEtlService {
.append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()))
.append(" (");
columnsMap
.forEach((targetColumnName, srcColumnName) -> insertSql.append(backtick).append(targetColumnName).append(backtick).append(","));
.forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(","));
int len = insertSql.length();
insertSql.delete(len - 1, len).append(") VALUES (");
@@ -102,7 +101,6 @@ public class RdbEtlService extends AbstractEtlService {
}
len = insertSql.length();
insertSql.delete(len - 1, len).append(")");
logger.info("executeSqlImport sql:{}",insertSql.toString());
try (Connection connTarget = targetDS.getConnection();
PreparedStatement pstmt = connTarget.prepareStatement(insertSql.toString())) {
connTarget.setAutoCommit(false);
@@ -116,7 +114,7 @@ public class RdbEtlService extends AbstractEtlService {
Map<String, Object> pkVal = new LinkedHashMap<>();
StringBuilder deleteSql = new StringBuilder(
"DELETE FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()) + " WHERE ");
appendCondition(dbMapping, deleteSql, pkVal, rs, backtick);
appendCondition(dbMapping, deleteSql, pkVal, rs);
try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) {
int k = 1;
for (Object val : pkVal.values()) {
@@ -134,6 +132,7 @@ public class RdbEtlService extends AbstractEtlService {
}
Integer type = columnType.get(targetClolumnName.toLowerCase());
Object value = rs.getObject(srcColumnName);
if (value != null) {
SyncUtil.setPStmt(type, pstmt, value, i);
@@ -181,7 +180,7 @@ public class RdbEtlService extends AbstractEtlService {
* 拼接目标表主键where条件
*/
private static void appendCondition(DbMapping dbMapping, StringBuilder sql, Map<String, Object> values,
ResultSet rs, String backtick) throws SQLException {
ResultSet rs) throws SQLException {
// 拼接主键
for (Map.Entry<String, String> entry : dbMapping.getTargetPk().entrySet()) {
String targetColumnName = entry.getKey();
@@ -189,7 +188,7 @@ public class RdbEtlService extends AbstractEtlService {
if (srcColumnName == null) {
srcColumnName = targetColumnName;
}
sql.append(backtick).append(targetColumnName).append(backtick).append("=? AND ");
sql.append(targetColumnName).append("=? AND ");
values.put(targetColumnName, rs.getObject(srcColumnName));
}
int len = sql.length();