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 e83aeb41af [Test][E2E] Stabilize schema evolution add-column
assertions (#11154)
e83aeb41af is described below
commit e83aeb41af14b4c4afd2b1a879a824c627fd9fdd
Author: Daniel <[email protected]>
AuthorDate: Mon Jun 22 17:18:43 2026 +0800
[Test][E2E] Stabilize schema evolution add-column assertions (#11154)
---
.../jdbc/AbstractSchemaChangeBaseIT.java | 78 ++++++++--------------
.../src/test/resources/ddl/add_columns.sql | 7 +-
2 files changed, 33 insertions(+), 52 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 f95554f79d..4448e6e906 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
@@ -309,48 +309,8 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
// case1 add columns with cdc data at same time
sourceDatabase.setTemplateName("add_columns").createAndInitialize();
- await().atMost(120, TimeUnit.SECONDS)
- .untilAsserted(
- () ->
- Assertions.assertIterableEquals(
- querySource(
- String.format(
- SOURCE_QUERY_COLUMNS,
- SOURCE_DATABASE,
- sourceTable)),
- querySink(
- String.format(
-
schemaChangeCase.getSinkQueryColumns(),
-
schemaChangeCase.getSchemaName(),
- sinkTable))));
- await().atMost(120, TimeUnit.SECONDS)
- .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));
-
- Assertions.assertIterableEquals(
- querySource(
- String.format(
- PROJECTION_QUERY,
- SOURCE_DATABASE,
- sourceTable)),
- querySink(
- String.format(
- PROJECTION_QUERY,
-
schemaChangeCase.getSchemaName(),
- sinkTable)
- + ORDER_BY));
- });
+ waitForSinkColumnsCatchUp(sourceTable, sinkTable);
+ assertAddColumnsDataSynced(sourceTable, sinkTable);
// case2 drop columns with cdc data at same time
assertCaseByDdlName("drop_columns");
@@ -383,21 +343,37 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
// case1 add columns with cdc data at same time
sourceDatabase.setTemplateName("add_columns").createAndInitialize();
- given().pollDelay(Duration.ofSeconds(5))
- .await()
- .atMost(SCHEMA_ASSERT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)
+ waitForSinkColumnsCatchUp(sourceTable, sinkTable);
+ assertAddColumnsDataSynced(sourceTable, 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(
() ->
Assertions.assertIterableEquals(
querySource(
- String.format(QUERY,
SOURCE_DATABASE, sourceTable)),
+ String.format(
+ SOURCE_QUERY_COLUMNS,
+ SOURCE_DATABASE,
+ sourceTable)),
querySink(
String.format(
- QUERY,
-
schemaChangeCase.getSchemaName(),
- sinkTable)
- + ORDER_BY)));
- await().atMost(120, TimeUnit.SECONDS)
+
schemaChangeCase.getSinkQueryColumns(),
+
schemaChangeCase.getSchemaName(),
+ sinkTable))));
+ }
+
+ /**
+ * Validates both the new add-columns rows and the projected full-table
view once schema
+ * evolution has settled on the sink side.
+ */
+ private void assertAddColumnsDataSynced(String sourceTable, String
sinkTable) {
+ await().atMost(SCHEMA_ASSERT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)
.untilAsserted(
() -> {
Assertions.assertIterableEquals(
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 a56744de00..dc334b7b73 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
@@ -34,6 +34,8 @@ 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);
update products set name = 'hawk9821' where id = 110;
insert into products
@@ -51,6 +53,8 @@ delete from products where id = 118;
alter table products ADD COLUMN add_column3 float not null default 1.1;
## 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);
delete from products where id = 113;
insert into products
@@ -66,6 +70,8 @@ values (128,"scooter","Small 2-wheel
scooter",3.14,'xx',1,1.1,'2023-02-02 09:09:
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);
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'),
@@ -80,4 +86,3 @@ values (173,"scooter","Small 2-wheel
scooter",3.14,'xx',1,1.1,'2023-02-02 09:09:
-- add column for irrelevant table
ALTER TABLE products_on_hand ADD COLUMN add_column5 varchar(64) not null
default 'yy';
-
