This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 1827600a2b [Test][E2E] Stabilize JDBC schema change assertions (#11200)
1827600a2b is described below
commit 1827600a2b9c99390d3fc13fc91babff537f6c1a
Author: Daniel <[email protected]>
AuthorDate: Tue Jun 30 23:02:56 2026 +0800
[Test][E2E] Stabilize JDBC schema change assertions (#11200)
---
.../jdbc/AbstractSchemaChangeBaseIT.java | 218 +++++++++++++++------
.../src/test/resources/ddl/add_columns.sql | 17 +-
2 files changed, 168 insertions(+), 67 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-ddl/src/test/java/org/apache/seatunnel/connectors/jdbc/AbstractSchemaChangeBaseIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-ddl/src/test/java/org/apache/seatunnel/connectors/jdbc/AbstractSchemaChangeBaseIT.java
index 99d1826937..8a7dcd8a35 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-ddl/src/test/java/org/apache/seatunnel/connectors/jdbc/AbstractSchemaChangeBaseIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-ddl/src/test/java/org/apache/seatunnel/connectors/jdbc/AbstractSchemaChangeBaseIT.java
@@ -55,6 +55,7 @@ import java.sql.DriverManager;
import java.sql.NClob;
import java.sql.ResultSet;
import java.sql.SQLException;
+import java.sql.Statement;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
@@ -77,6 +78,23 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
private static final String SOURCE_DATABASE = "shop";
private static final String SOURCE_TABLE = "products";
+ /**
+ * Deterministic source row used to prove the MySQL CDC reader is
consuming binlog events before
+ * schema-change DDL is executed.
+ */
+ private static final int STREAM_READY_MARKER_ID = 1000;
+ /**
+ * Marker value written to the source table so sink-side readiness polling
can identify the
+ * probe row without depending on connector internals.
+ */
+ private static final String STREAM_READY_MARKER_NAME =
"__cdc_stream_ready__";
+ /**
+ * Stable payload for the readiness probe row; keeping it constant makes
repeated test attempts
+ * idempotent through the upsert statement.
+ */
+ private static final String STREAM_READY_MARKER_DESCRIPTION =
+ "wait for binlog stream readiness";
+
private static final String MYSQL_HOST = "mysql_cdc_e2e";
private static final String MYSQL_USER_NAME = "mysqluser";
private static final String MYSQL_USER_PASSWORD = "mysqlpw";
@@ -85,6 +103,7 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
private static final String QUERY = "select * from %s.%s";
private static final String PROJECTION_QUERY =
"select
id,name,description,weight,add_column1,add_column2,add_column3 from %s.%s";
+ private static final String SOURCE_DESC_QUERY = "desc %s.%s";
private static final String SOURCE_QUERY_COLUMNS =
"SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE
TABLE_SCHEMA = '%s' AND TABLE_NAME = '%s' ORDER by COLUMN_NAME";
@@ -294,18 +313,16 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
}
private void assertSchemaEvolution(String sourceTable, String sinkTable) {
- await().atMost(120, TimeUnit.SECONDS)
+ // The exactly-once path can report RUNNING before the sink finishes
the first XA batch in
+ // slower CI environments, so reuse the longer schema assertion
timeout for the initial
+ // data catch-up instead of failing on a transient empty sink table.
+ await().atMost(SCHEMA_ASSERT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)
.untilAsserted(
() ->
- Assertions.assertIterableEquals(
- querySource(
- String.format(QUERY,
SOURCE_DATABASE, sourceTable)),
- querySink(
- String.format(
- QUERY,
-
schemaChangeCase.getSchemaName(),
- sinkTable)
- + ORDER_BY)));
+ assertTableDataEqualsBySourceColumnOrder(
+ sourceTable, sinkTable, null));
+
+ waitForStreamingReady(sourceTable, sinkTable);
// case1 add columns with cdc data at same time
sourceDatabase.setTemplateName("add_columns").createAndInitialize();
@@ -322,24 +339,19 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
assertCaseByDdlName("modify_columns");
}
- private void assertCaseByDdlName(String drop_columns) {
- sourceDatabase.setTemplateName(drop_columns).createAndInitialize();
+ private void assertCaseByDdlName(String ddlTemplateName) {
+ sourceDatabase.setTemplateName(ddlTemplateName).createAndInitialize();
assertTableStructureAndData(SOURCE_TABLE,
schemaChangeCase.getSinkTable2());
}
private void assertSchemaEvolutionForAddColumns(String sourceTable, String
sinkTable) {
- await().atMost(120, TimeUnit.SECONDS)
+ await().atMost(SCHEMA_ASSERT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)
.untilAsserted(
() ->
- Assertions.assertIterableEquals(
- querySource(
- String.format(QUERY,
SOURCE_DATABASE, sourceTable)),
- querySink(
- String.format(
- QUERY,
-
schemaChangeCase.getSchemaName(),
- sinkTable)
- + ORDER_BY)));
+ assertTableDataEqualsBySourceColumnOrder(
+ sourceTable, sinkTable, null));
+
+ waitForStreamingReady(sourceTable, sinkTable);
// case1 add columns with cdc data at same time
sourceDatabase.setTemplateName("add_columns").createAndInitialize();
@@ -348,26 +360,51 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
}
/**
- * Schema-change sinks can publish the new rows before the sink table
metadata is fully updated.
- * Waiting for the column list first avoids racing the add-columns data
assertions.
+ * Snapshot convergence does not prove the MySQL CDC reader has entered
steady-state binlog
+ * consumption. Emit one deterministic DML event and wait until the sink
receives it before
+ * running schema-change DDL bursts.
*/
- private void waitForSinkColumnsCatchUp(String sourceTable, String
sinkTable) {
+ private void waitForStreamingReady(String sourceTable, String sinkTable) {
+ executeSourceSql(
+ String.format(
+ "INSERT INTO %s.%s (id, name, description, weight) "
+ + "VALUES (%d, '%s', '%s', 0.0) "
+ + "ON DUPLICATE KEY UPDATE "
+ + "name = VALUES(name), description =
VALUES(description), weight = VALUES(weight)",
+ SOURCE_DATABASE,
+ sourceTable,
+ STREAM_READY_MARKER_ID,
+ STREAM_READY_MARKER_NAME,
+ STREAM_READY_MARKER_DESCRIPTION));
+
+ String readyQuery =
+ String.format(
+ "select id,name,description,weight from %%s.%%s where
id = %d order by id",
+ STREAM_READY_MARKER_ID);
await().atMost(SCHEMA_ASSERT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)
.untilAsserted(
() ->
Assertions.assertIterableEquals(
querySource(
String.format(
- SOURCE_QUERY_COLUMNS,
- SOURCE_DATABASE,
- sourceTable)),
+ readyQuery,
SOURCE_DATABASE, sourceTable)),
querySink(
String.format(
-
schemaChangeCase.getSinkQueryColumns(),
+ readyQuery,
schemaChangeCase.getSchemaName(),
sinkTable))));
}
+ /**
+ * Schema-change sinks can publish the new rows before the sink table
metadata is fully updated.
+ * Waiting for the column list first avoids racing the add-columns data
assertions.
+ */
+ private void waitForSinkColumnsCatchUp(String sourceTable, String
sinkTable) {
+ await().atMost(SCHEMA_ASSERT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
assertColumnNamesEqualsIgnoringPhysicalOrder(sourceTable, sinkTable));
+ }
+
/**
* Validates both the new add-columns rows and the projected full-table
view once schema
* evolution has settled on the sink side.
@@ -376,17 +413,8 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
await().atMost(SCHEMA_ASSERT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)
.untilAsserted(
() -> {
- Assertions.assertIterableEquals(
- querySource(
- String.format(QUERY,
SOURCE_DATABASE, sourceTable)
- + " where id >= 128"),
- querySink(
- String.format(
- QUERY,
-
schemaChangeCase.getSchemaName(),
- sinkTable)
- + " where id >= 128"
- + ORDER_BY));
+ assertTableDataEqualsBySourceColumnOrder(
+ sourceTable, sinkTable, "id >= 128");
Assertions.assertIterableEquals(
querySource(
@@ -408,30 +436,85 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
.await()
.atMost(SCHEMA_ASSERT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)
.untilAsserted(
- () ->
- Assertions.assertIterableEquals(
- querySource(
- String.format(
- SOURCE_QUERY_COLUMNS,
- SOURCE_DATABASE,
- sourceTable)),
- querySink(
- String.format(
-
schemaChangeCase.getSinkQueryColumns(),
-
schemaChangeCase.getSchemaName(),
- sinkTable))));
+ () ->
assertColumnNamesEqualsIgnoringPhysicalOrder(sourceTable, sinkTable));
await().atMost(SCHEMA_ASSERT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)
.untilAsserted(
() ->
- Assertions.assertIterableEquals(
- querySource(
- String.format(QUERY,
SOURCE_DATABASE, sourceTable)),
- querySink(
- String.format(
- QUERY,
-
schemaChangeCase.getSchemaName(),
- sinkTable)
- + ORDER_BY)));
+ assertTableDataEqualsBySourceColumnOrder(
+ sourceTable, sinkTable, null));
+ }
+
+ /**
+ * JDBC schema evolution can keep the effective column set while
materializing a different
+ * physical order in the sink, so schema assertions should compare
normalized column names.
+ */
+ private void assertColumnNamesEqualsIgnoringPhysicalOrder(
+ String sourceTable, String sinkTable) {
+ Assertions.assertIterableEquals(
+ normalizeColumnNames(
+ querySource(
+ String.format(SOURCE_QUERY_COLUMNS,
SOURCE_DATABASE, sourceTable))),
+ normalizeColumnNames(
+ querySink(
+ String.format(
+ schemaChangeCase.getSinkQueryColumns(),
+ schemaChangeCase.getSchemaName(),
+ sinkTable))));
+ }
+
+ /**
+ * Projects sink data with the current source column order so row
assertions stay stable when a
+ * JDBC sink reorders equivalent columns after applying schema changes.
+ */
+ private void assertTableDataEqualsBySourceColumnOrder(
+ String sourceTable, String sinkTable, String whereClause) {
+ List<String> sourceColumns = getSourceColumnNames(sourceTable);
+ Assertions.assertIterableEquals(
+ querySource(
+ buildProjectionQuery(
+ SOURCE_DATABASE, sourceTable, sourceColumns,
whereClause)),
+ querySink(
+ buildProjectionQuery(
+ schemaChangeCase.getSchemaName(),
+ sinkTable,
+ sourceColumns,
+ whereClause)));
+ }
+
+ /** Reads the current MySQL source schema order that downstream row
assertions should follow. */
+ private List<String> getSourceColumnNames(String sourceTable) {
+ List<String> sourceColumns = new ArrayList<>();
+ for (List<Object> row :
+ querySource(String.format(SOURCE_DESC_QUERY, SOURCE_DATABASE,
sourceTable))) {
+ sourceColumns.add(String.valueOf(row.get(0)));
+ }
+ return sourceColumns;
+ }
+
+ /** Builds a deterministic projection query without relying on
sink-specific physical order. */
+ private String buildProjectionQuery(
+ String database, String table, List<String> columns, String
whereClause) {
+ StringBuilder queryBuilder =
+ new StringBuilder("select ")
+ .append(String.join(",", columns))
+ .append(" from ")
+ .append(database)
+ .append(".")
+ .append(table);
+ if (StringUtils.isNotBlank(whereClause)) {
+ queryBuilder.append(" where ").append(whereClause);
+ }
+ return queryBuilder.append(ORDER_BY).toString();
+ }
+
+ /** Sorts schema query output by column name so assertions ignore
placement-only differences. */
+ private List<String> normalizeColumnNames(List<List<Object>> rows) {
+ List<String> normalizedColumnNames = new ArrayList<>();
+ for (List<Object> row : rows) {
+ normalizedColumnNames.add(String.valueOf(row.get(0)));
+ }
+ normalizedColumnNames.sort(String::compareTo);
+ return normalizedColumnNames;
}
private Connection getJdbcConnection(String connectionType) throws
SQLException {
@@ -470,6 +553,19 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
}
}
+ /**
+ * Executes a source-side DML statement directly against MySQL to produce
a CDC event that the
+ * running SeaTunnel job must consume.
+ */
+ private void executeSourceSql(String sql) {
+ try (Connection connection = getJdbcConnection("source");
+ Statement statement = connection.createStatement()) {
+ statement.execute(sql);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
private List<List<Object>> querySink(String sql) {
try (Connection connection = getJdbcConnection("sink")) {
ResultSet resultSet =
connection.createStatement().executeQuery(sql);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-ddl/src/test/resources/ddl/add_columns.sql
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-ddl/src/test/resources/ddl/add_columns.sql
index dc334b7b73..a213c20730 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-ddl/src/test/resources/ddl/add_columns.sql
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-ddl/src/test/resources/ddl/add_columns.sql
@@ -33,9 +33,12 @@ VALUES (110,"scooter","Small 2-wheel scooter",3.14),
update products set name = 'hawk9821' where id = 101;
delete from products where id = 102;
-alter table products ADD COLUMN add_column1 varchar(64) not null default
'yy',ADD COLUMN add_column2 int not null default 1;
--- Let the source observe the DDL event before the first row uses the new
columns.
-DO SLEEP(5);
+alter table products ADD COLUMN add_column1 varchar(64) not null default 'yy';
+-- Debezium updates schema history asynchronously, so keep adjacent DDLs in
separate binlog batches.
+DO SLEEP(10);
+alter table products ADD COLUMN add_column2 int not null default 1;
+-- Let the source observe both DDL events before the first row uses the new
columns.
+DO SLEEP(15);
update products set name = 'hawk9821' where id = 110;
insert into products
@@ -51,10 +54,12 @@ values (119,"scooter","Small 2-wheel scooter",3.14,'xx',1),
delete from products where id = 118;
alter table products ADD COLUMN add_column3 float not null default 1.1;
+-- Keep adjacent add-column DDLs in separate schema-history batches.
+DO SLEEP(10);
## timestamp is not supported as a cross-database default values for DDL
statements
alter table products ADD COLUMN add_column4 timestamp;
--- The second add-columns batch also needs a short gap before the new-column
DML arrives.
-DO SLEEP(5);
+-- The second add-columns batch also needs a longer gap before the new-column
DML arrives.
+DO SLEEP(15);
delete from products where id = 113;
insert into products
@@ -71,7 +76,7 @@ update products set name = 'hawk9821' where id = 135;
alter table products ADD COLUMN add_column6 varchar(64) not null default 'ff';
-- Keep the final add-column DDL and the follow-up DML in separate CDC batches.
-DO SLEEP(5);
+DO SLEEP(15);
delete from products where id = 115;
insert into products
values (173,"scooter","Small 2-wheel scooter",3.14,'xx',1,1.1,'2023-02-02
09:09:09','tt'),
