org.mockito
@@ -289,6 +297,65 @@
+
+ org.codehaus.mojo
+ build-helper-maven-plugin
+ 3.2.0
+
+
+ add-integration-test-source
+ generate-test-sources
+
+ add-test-source
+ add-test-resource
+
+
+
+ src/it/java
+
+
+
+ src/it/resources
+
+
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-failsafe-plugin
+ 3.5.6
+
+ ${skipITs}
+
+
+ -Xmx4g
+ -Dapp_type=DAAS
+ --add-opens=java.base/java.lang=ALL-UNNAMED
+ --add-opens=java.base/java.lang.reflect=ALL-UNNAMED
+ --add-opens=java.base/java.util=ALL-UNNAMED
+ --add-opens=java.base/java.security=ALL-UNNAMED
+ --add-opens=java.base/java.io=ALL-UNNAMED
+ --add-opens=java.base/java.time=ALL-UNNAMED
+ --add-opens=java.base/jdk.internal.loader=ALL-UNNAMED
+
+
+ ${project.build.directory}/failsafe-reports
+
+
+
+
+ integration-test
+
+ integration-test
+ verify
+
+
+
+
diff --git a/connectors/mongodb-connector/src/it/java/io/tapdata/mongodb/MongoDBConnectorIT.java b/connectors/mongodb-connector/src/it/java/io/tapdata/mongodb/MongoDBConnectorIT.java
new file mode 100644
index 000000000..9e4b1b7e6
--- /dev/null
+++ b/connectors/mongodb-connector/src/it/java/io/tapdata/mongodb/MongoDBConnectorIT.java
@@ -0,0 +1,139 @@
+package io.tapdata.mongodb;
+
+import io.tapdata.entity.codec.TapCodecsRegistry;
+import io.tapdata.entity.logger.TapLog;
+import io.tapdata.entity.utils.DataMap;
+import io.tapdata.it.ConnectorIT;
+import io.tapdata.it.ConnectorTestContext;
+import io.tapdata.it.support.TestStateMap;
+import io.tapdata.pdk.apis.context.TapConnectorContext;
+import io.tapdata.pdk.apis.functions.ConnectorFunctions;
+import org.bson.BsonRegularExpression;
+import org.bson.BsonTimestamp;
+import org.bson.types.Binary;
+import org.bson.types.Code;
+import org.bson.types.Decimal128;
+import org.bson.types.ObjectId;
+import org.bson.types.Symbol;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * MongoDB 连接器通用集成测试。
+ *
+ * 继承 {@link ConnectorIT} 后自动运行框架内置的全部通用集成测试用例
+ * （连接元数据、集合 DDL、数据读写、事务、流式读取、命令等），
+ * 对 MongoDB 不支持的 ConnectorFunctions 能力自动跳过。
+ * 必实现能力由 {@link #requiredCapabilities()} 主动声明（原则 3：声明式能力），
+ * 框架据此校验：声明必实现但未实现、或已实现但无用例覆盖 → 测试失败。
+ *
+ * 连接配置读取 src/it/resources/config/mongodb-connection.json
+ * （uri/database），支持系统属性/环境变量覆盖
+ * （如 -Dconnector.it.uri=xxx 或 CONNECTOR_IT_URI=xxx）。
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class MongoDBConnectorIT extends ConnectorIT {
+
+ @Override
+ protected Set requiredCapabilities() {
+ // MongoDB 同时承担源与目标角色：声明对外承诺的全部能力（与 registerCapabilities 注册保持一致；
+ // memoryFetcher 为引擎诊断钩子，由基类 ignoredCapabilities() 排除，不参与校验）
+ return Stream.of("connectionTest", "discoverSchema", "getTableNames",
+ "createTableV2", "dropTable", "batchCount", "batchRead", "streamRead",
+ "timestampToStreamOffset", "queryByAdvanceFilter", "countByPartitionFilter",
+ "writeRecord", "createIndex", "queryIndexes", "errorHandle",
+ "executeCommand", "getTableInfo", "getReadPartitions", "queryFieldMinMaxValue",
+ "transactionBegin", "transactionCommit", "transactionRollback").collect(Collectors.toSet());
+ }
+
+ @Override
+ protected ConnectorTestContext createContext() throws Throwable {
+ // 1. 创建连接器
+ MongodbConnector connector = new MongodbConnector();
+
+ // 2. 连接配置（URI 模式，与产品连接表单默认一致）
+ DataMap config = readConnectionConfig("config/mongodb-connection.json");
+
+ // 3. 构建 NodeContext：specification + connectionConfig + nodeConfig + log
+ // specification 提供 dataTypesMap，discoverSchema 的 tapType 自动填充依赖它
+ TapConnectorContext nodeContext = new TapConnectorContext(
+ loadSpecification("spec.json"), config, DataMap.create(), new TapLog());
+ nodeContext.setStateMap(new TestStateMap());
+
+ // 4. 注册能力（registerCapabilities 产物，基类据此动态检测并驱动用例）
+ ConnectorFunctions functions = new ConnectorFunctions();
+ TapCodecsRegistry codecRegistry = TapCodecsRegistry.create();
+ connector.registerCapabilities(functions, codecRegistry);
+
+ // 5. 组装测试上下文（特性开关：MongoDB NoSQL 特性适配）
+ ConnectorTestContext ctx = ConnectorTestContext.builder()
+ .connector(connector)
+ .nodeContext(nodeContext)
+ .connectorFunctions(functions)
+ .codecRegistry(codecRegistry)
+ .config(config)
+ .log(new TapLog())
+ // MongoDB createTableV2 幂等建集合固定返回 tableExists=false
+ .createTableReportsTableExists(false)
+ // schema-free：空集合无法推断字段，需先写入采样数据再 discoverSchema
+ .schemaDiscoveryRequiresSampleData(true)
+ // discoverSchema 会带出隐式 _id 字段
+ .schemaAllowsExtraFields(true)
+ // 主键为库自动生成的 _id，业务字段不被标记为主键
+ .schemaPrimaryKeyStrict(false)
+ // executeCommand 仅支持 execute/executeQuery/count/aggregate，不支持 ping
+ .executeCommandSupportsPing(false)
+ // queryFieldMinMaxValue 基于分区索引字段计算 min/max
+ .fieldMinMaxRequiresPartitionIndex(true)
+ .build();
+ ctx.getLog().info("[IT] MongoDB connection: uri={}, database={}",
+ config.getString("uri"), config.getString("database"));
+ return ctx;
+ }
+
+ @Override
+ protected List> beforeWrite(List> rows) {
+ // unique 索引用例（idx_c_int）要求 c_int 唯一：随机值冲突时改为超出随机范围的序列值
+ Set seen = new HashSet<>();
+ long seq = 1_000_001L;
+ for (Map row : rows) {
+ long v = ((Number) row.get("c_int")).longValue();
+ if (!seen.add(v)) {
+ row.put("c_int", seq++);
+ }
+ }
+ return rows;
+ }
+
+ @Override
+ protected Map specialValueSamples() {
+ // MongoDB 特有 BSON 类型样本：须与 registerCapabilities 注册的 ToTapValueCodec 一一对应
+ // （U8 用例验证这些特殊值 wrap 后被连接器 codec 识别，而非 TapRawValue 兜底）
+ Map samples = new LinkedHashMap<>();
+ samples.put("obj_objectid", new ObjectId());
+ samples.put("obj_binary", new Binary((byte) 0x80, new byte[]{1, 2, 3}));
+ samples.put("obj_code", new Code("function() { return 1; }"));
+ samples.put("obj_decimal128", Decimal128.parse("12345.6789"));
+ samples.put("obj_symbol", new Symbol("sym"));
+ samples.put("obj_bson_timestamp", new BsonTimestamp(1_700_000_000, 1));
+ samples.put("obj_regex", new BsonRegularExpression("^tap.*", "i"));
+ return samples;
+ }
+
+ @DisplayName("Test github action")
+ @Test
+ public void testGithubAction() {
+ int result = 10*7;
+ Assertions.assertEquals(70, result);
+ }
+}
diff --git a/connectors/mongodb-connector/src/it/resources/config/mongodb-connection.json b/connectors/mongodb-connector/src/it/resources/config/mongodb-connection.json
new file mode 100644
index 000000000..b483c9502
--- /dev/null
+++ b/connectors/mongodb-connector/src/it/resources/config/mongodb-connection.json
@@ -0,0 +1,4 @@
+{
+ "uri": "",
+ "database": ""
+}
diff --git a/connectors/mysql-connector/pom.xml b/connectors/mysql-connector/pom.xml
index 4b36c0eb2..787a82546 100644
--- a/connectors/mysql-connector/pom.xml
+++ b/connectors/mysql-connector/pom.xml
@@ -22,8 +22,9 @@
- 8
+ 17
2.0.5-SNAPSHOT
+ true
@@ -54,11 +55,26 @@
5.8.1
test
+
+
+ io.tapdata
+ tapdata-connector-it
+ 1.0-SNAPSHOT
+ test
+
io.tapdata
mysql-core
1.0-SNAPSHOT
+
+
+ io.tapdata
+ tapdata-common
+ 2.9-SNAPSHOT
+ test
+
io.tapdata
pdk-error-code
@@ -209,6 +225,24 @@
+
+ add-integration-test-source
+ generate-test-sources
+
+ add-test-source
+ add-test-resource
+
+
+
+ src/it/java
+
+
+
+ src/it/resources
+
+
+
+
@@ -232,7 +266,40 @@
-
+
+
+ org.apache.maven.plugins
+ maven-failsafe-plugin
+ 3.5.6
+
+ ${skipITs}
+
+
+ -Xmx8g
+ --add-opens=java.base/java.lang=ALL-UNNAMED
+ --add-opens=java.base/java.lang.reflect=ALL-UNNAMED
+ --add-opens=java.base/java.util=ALL-UNNAMED
+ --add-opens=java.base/java.security=ALL-UNNAMED
+ --add-opens=java.base/java.io=ALL-UNNAMED
+ --add-opens=java.base/java.time=ALL-UNNAMED
+ --add-opens=java.base/jdk.internal.loader=ALL-UNNAMED
+
+
+ ${project.build.directory}/failsafe-reports
+
+
+
+
+ integration-test
+
+ integration-test
+ verify
+
+
+
+
+
diff --git a/connectors/mysql-connector/src/it/java/io/tapdata/connector/mysql/MySQLConnectorIT.java b/connectors/mysql-connector/src/it/java/io/tapdata/connector/mysql/MySQLConnectorIT.java
new file mode 100644
index 000000000..2a0142032
--- /dev/null
+++ b/connectors/mysql-connector/src/it/java/io/tapdata/connector/mysql/MySQLConnectorIT.java
@@ -0,0 +1,89 @@
+package io.tapdata.connector.mysql;
+
+import io.tapdata.entity.codec.TapCodecsRegistry;
+import io.tapdata.entity.logger.TapLog;
+import io.tapdata.entity.utils.DataMap;
+import io.tapdata.it.ConnectorIT;
+import io.tapdata.it.ConnectorTestContext;
+import io.tapdata.it.support.TestStateMap;
+import io.tapdata.pdk.apis.context.TapConnectorContext;
+import io.tapdata.pdk.apis.functions.ConnectorFunctions;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * MySQL 连接器通用集成测试。
+ *
+ * 继承 {@link ConnectorIT} 后自动运行框架内置的全部通用集成测试用例
+ * （连接元数据、表 DDL、数据读写、事务、流式读取、命令等），
+ * 对 MySQL 不支持的 ConnectorFunctions 能力自动跳过。
+ * 必实现能力由 {@link #requiredCapabilities()} 主动声明（原则 3：声明式能力），
+ * 框架据此校验：声明必实现但未实现、或已实现但无用例覆盖 → 测试失败。
+ *
+ * 连接配置读取 src/it/resources/config/connection.json
+ * （host/port/database/user/password），支持系统属性/环境变量覆盖
+ * （如 -Dconnector.it.host=xxx 或 CONNECTOR_IT_HOST=xxx）。
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class MySQLConnectorIT extends ConnectorIT {
+
+ @Override
+ protected Set requiredCapabilities() {
+ // MySQL 同时承担源与目标角色：声明对外承诺的全部能力。
+ // 声明后若实现被移除或框架无对应用例，原则 3 校验用例将失败（不遗漏必实现接口、不漏测已实现能力）
+ return Stream.of("connectionTest", "discoverSchema", "getTableNames",
+ "createTableV2", "dropTable", "clearTable", "batchCount", "batchRead",
+ "streamRead", "streamReadMultiConnection", "timestampToStreamOffset",
+ "queryByAdvanceFilter", "countByPartitionFilter", "writeRecord", "afterInitialSync",
+ "createIndex", "queryIndexes", "createConstraint", "queryConstraints", "dropConstraint",
+ "newField", "alterFieldName", "alterFieldAttributes", "dropField",
+ "errorHandle", "executeCommand", "executeCommandV2", "getTableInfo",
+ "runRawCommand", "transactionBegin", "transactionCommit", "transactionRollback",
+ "queryHashByAdvanceFilter", "exportEventSql").collect(Collectors.toSet());
+ }
+
+ @Override
+ protected ConnectorTestContext createContext() throws Throwable {
+ // 1. 创建连接器
+ MysqlConnector connector = new MysqlConnector();
+
+ // 2. 连接配置（与 src/test/resources 的占位 connection.json 隔离，避免资源覆盖冲突）
+ DataMap config = readConnectionConfig("config/mysql-connection.json");
+
+ // 3. 构建 NodeContext：specification + connectionConfig + nodeConfig + log
+ TapConnectorContext nodeContext = new TapConnectorContext(
+ null, config, DataMap.create().kv("enableTransaction", true), new TapLog());
+ nodeContext.setStateMap(new TestStateMap());
+
+ // 4. 注册能力（registerCapabilities 产物，基类据此动态检测并驱动用例）
+ ConnectorFunctions functions = new ConnectorFunctions();
+ TapCodecsRegistry codecRegistry = TapCodecsRegistry.create();
+ connector.registerCapabilities(functions, codecRegistry);
+
+ // 5. 组装测试上下文
+ ConnectorTestContext ctx = ConnectorTestContext.builder()
+ .connector(connector)
+ .nodeContext(nodeContext)
+ .connectorFunctions(functions)
+ .codecRegistry(codecRegistry)
+ .config(config)
+ .log(new TapLog())
+ .build();
+ ctx.getLog().info("[IT] MySQL connection: host={}, port={}, database={}, user={}",
+ config.getString("host"), config.getInteger("port"), config.getString("database"), config.getString("user"));
+ return ctx;
+ }
+
+ @DisplayName("Test github action")
+ @Test
+ public void testGithubAction() {
+ int result = 10*7;
+ Assertions.assertEquals(70, result);
+ }
+}
diff --git a/connectors/mysql-connector/src/it/resources/config/mysql-connection.json b/connectors/mysql-connector/src/it/resources/config/mysql-connection.json
new file mode 100644
index 000000000..5f10f8799
--- /dev/null
+++ b/connectors/mysql-connector/src/it/resources/config/mysql-connection.json
@@ -0,0 +1,8 @@
+{
+ "host": "",
+ "port": 3306,
+ "database": "",
+ "user": "",
+ "password": "",
+ "highPerformance": true
+}
diff --git a/connectors/mysql-connector/src/main/java/io/tapdata/connector/mysql/MysqlConnector.java b/connectors/mysql-connector/src/main/java/io/tapdata/connector/mysql/MysqlConnector.java
index 734547e69..4ac26e89d 100644
--- a/connectors/mysql-connector/src/main/java/io/tapdata/connector/mysql/MysqlConnector.java
+++ b/connectors/mysql-connector/src/main/java/io/tapdata/connector/mysql/MysqlConnector.java
@@ -773,7 +773,7 @@ public void buildIllegalDateFieldName(TapRecordEvent event, List illegal
@Override
protected String getBatchReadSelectSql(TapTable tapTable) {
if (tapTable.getNameFieldMap().size() > 50) {
- return String.format("SELECT * %s", getSchemaAndTable(tapTable.getId()));
+ return String.format("SELECT * FROM %s", getSchemaAndTable(tapTable.getId()));
} else {
return super.getBatchReadSelectSql(tapTable);
}
diff --git a/connectors/mysql-connector/src/test/java/io/tapdata/connector/mysql/MysqlConnectorTest.java b/connectors/mysql-connector/src/test/java/io/tapdata/connector/mysql/MysqlConnectorTest.java
index d2052169c..259cf5d36 100644
--- a/connectors/mysql-connector/src/test/java/io/tapdata/connector/mysql/MysqlConnectorTest.java
+++ b/connectors/mysql-connector/src/test/java/io/tapdata/connector/mysql/MysqlConnectorTest.java
@@ -235,15 +235,17 @@ void test7() throws SQLException {
class BatchReadSqlTest {
TapTable tapTable;
MysqlConfig mysqlConfig;
+ CommonDbConfig commonDbConfig;
MysqlConnector connector;
@BeforeEach
void setUp() {
tapTable = mock(TapTable.class);
mysqlConfig = mock(MysqlConfig.class);
- connector = mock(MysqlConnector.class);
- doCallRealMethod().when(connector).getBatchReadSelectSql(tapTable);
+ commonDbConfig = mock(CommonDbConfig.class);
+ connector = new MysqlConnector();
UnitTestUtils.injectField(MysqlConnector.class, connector, "mysqlConfig", mysqlConfig);
+ UnitTestUtils.injectField(CommonDbConnector.class, connector, "commonDbConfig", commonDbConfig);
}
private LinkedHashMap generateFieldMap(TapField... fields) {
@@ -265,7 +267,11 @@ void testFieldSizeLargeThen50() {
fields[i] = new TapField("f" + i, "INT");
}
when(tapTable.getNameFieldMap()).thenReturn(generateFieldMap(fields));
- assertTrue(connector.getBatchReadSelectSql(tapTable).toLowerCase().startsWith("select *"));
+ when(tapTable.getId()).thenReturn("AA_0607");
+ when(commonDbConfig.getEscapeChar()).thenReturn('`');
+ when(commonDbConfig.getSchema()).thenReturn("COOLGJ");
+
+ assertEquals("SELECT * FROM `COOLGJ`.`AA_0607`", connector.getBatchReadSelectSql(tapTable));
}
}