This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12654-8c7440b4cb3ea0b55e98d764e5354b3bb6c1cd70 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit fc1448c48ff93a046292b495590124813838786c Author: Jast <[email protected]> AuthorDate: Thu Oct 8 13:23:26 2026 +0000 [Test][Connector-V2] Poll for CDC and RabbitMQ data readiness instead of fixed sleeps in e2e tests (#12654) Co-authored-by: root <[email protected]> --- .../cdc/oracle/OracleCDCWithSchemaChangeIT.java | 9 ++- .../e2e/connector/hudi/HudiSinkCDCIT.java | 15 ++--- .../e2e/connector/iceberg/IcebergSinkCDCIT.java | 71 ++++++++++------------ .../e2e/connector/rabbitmq/RabbitmqIT.java | 29 +++++++-- 4 files changed, 70 insertions(+), 54 deletions(-) diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCWithSchemaChangeIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCWithSchemaChangeIT.java index 0b8dbc1463..6e922d5546 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCWithSchemaChangeIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCWithSchemaChangeIT.java @@ -162,9 +162,9 @@ public class OracleCDCWithSchemaChangeIT extends AbstractOracleCDCIT implements } }); - // Waiting to job running for auto create sink table - Thread.sleep(10000L); - + // The sink table is auto-created once the job reaches RUNNING; the data wait + // inside assertSchemaEvolution tolerates the table not existing yet, so no + // fixed pre-wait is needed here. assertSchemaEvolution( ORACLE_CONTAINER.getJdbcUrl(), ORACLE_CONTAINER.getJdbcUrl(), @@ -214,7 +214,10 @@ public class OracleCDCWithSchemaChangeIT extends AbstractOracleCDCIT implements String sinkTableName, boolean oracle2Mysql) throws Exception { + // Tolerate transient SQL errors while the CDC job is still creating the + // sink table, instead of relying on a fixed pre-wait before this call. await().atMost(300, TimeUnit.SECONDS) + .ignoreExceptions() .untilAsserted( () -> checkData( diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-hudi-e2e/src/test/java/org/apache/seatunnel/e2e/connector/hudi/HudiSinkCDCIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-hudi-e2e/src/test/java/org/apache/seatunnel/e2e/connector/hudi/HudiSinkCDCIT.java index d4d571ed77..ddd751497a 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-hudi-e2e/src/test/java/org/apache/seatunnel/e2e/connector/hudi/HudiSinkCDCIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-hudi-e2e/src/test/java/org/apache/seatunnel/e2e/connector/hudi/HudiSinkCDCIT.java @@ -65,7 +65,6 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.stream.Stream; -import static java.lang.Thread.sleep; import static org.apache.seatunnel.e2e.common.container.AbstractTestContainer.HOST_VOLUME_MOUNT_PATH; import static org.awaitility.Awaitility.given; @@ -284,14 +283,15 @@ public class HudiSinkCDCIT extends TestSuiteBase implements TestResource { private void insertAndCheckData(TestContainer container) throws InterruptedException { // Init table data initSourceTableData(MYSQL_DATABASE, SOURCE_TABLE); - // Waiting 30s for source capture data - sleep(30000); Configuration configuration = new Configuration(); configuration.set("fs.defaultFS", LocalFileSystem.DEFAULT_FS); + // The wait below already polls for the CDC data, so the former fixed 30s + // pre-wait was folded into the budget (60s + 30s) instead of blocking + // unconditionally first. given().ignoreExceptions() .await() - .atMost(60000, TimeUnit.MILLISECONDS) + .atMost(90000, TimeUnit.MILLISECONDS) .untilAsserted( () -> { Path newestCommitFilePath = @@ -323,14 +323,15 @@ public class HudiSinkCDCIT extends TestSuiteBase implements TestResource { private void upsertAndCheckData(TestContainer container) throws InterruptedException, IOException { upsertDeleteSourceTable(MYSQL_DATABASE, SOURCE_TABLE); - // Waiting 30s for source capture data - sleep(30000); Configuration configuration = new Configuration(); configuration.set("fs.defaultFS", LocalFileSystem.DEFAULT_FS); + // The wait below already polls for the CDC data, so the former fixed 30s + // pre-wait was folded into the budget (60s + 30s) instead of blocking + // unconditionally first. given().ignoreExceptions() .await() - .atMost(60000, TimeUnit.MILLISECONDS) + .atMost(90000, TimeUnit.MILLISECONDS) .untilAsserted( () -> { Path newestCommitFilePath = diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iceberg/IcebergSinkCDCIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iceberg/IcebergSinkCDCIT.java index bdfe475c51..eb4e5ac270 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iceberg/IcebergSinkCDCIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iceberg/IcebergSinkCDCIT.java @@ -249,12 +249,13 @@ public class IcebergSinkCDCIT extends TestSuiteBase implements TestResource { MYSQL_DATABASE, SOURCE_TABLE, addField1, addField2, addField3); executeSql(insertMultiColumnSql); - sleep(30000); // Wait for source capture data - - // Verify that multiple columns were added and data is correct + // Verify that multiple columns were added and data is correct. + // The wait below already polls for the CDC data, so the former fixed 30s + // pre-wait was folded into the budget (120s + 30s) instead of blocking + // unconditionally first. given().ignoreExceptions() .await() - .atMost(120000, TimeUnit.MILLISECONDS) + .atMost(150000, TimeUnit.MILLISECONDS) .untilAsserted( () -> { Schema schema = loadIcebergSchema(); @@ -354,12 +355,12 @@ public class IcebergSinkCDCIT extends TestSuiteBase implements TestResource { MYSQL_DATABASE, SOURCE_TABLE, modifyTypeField1, modifyTypeField2); executeSql(insertAfterModifyTypesSql); - sleep(30000); // Wait for source capture data - - // Verify that column types were modified and data is correct + // Verify that column types were modified and data is correct. + // The wait below already polls for the CDC data; the former fixed 30s + // pre-wait was folded into the budget (120s + 30s). given().ignoreExceptions() .await() - .atMost(120000, TimeUnit.MILLISECONDS) + .atMost(150000, TimeUnit.MILLISECONDS) .untilAsserted( () -> { Schema schema = loadIcebergSchema(); @@ -455,12 +456,12 @@ public class IcebergSinkCDCIT extends TestSuiteBase implements TestResource { MYSQL_DATABASE, SOURCE_TABLE, modifyField1, modifyField2); executeSql(insertAfterModifySql); - sleep(30000); // Wait for source capture data - - // Verify that columns were modified and data is correct + // Verify that columns were modified and data is correct. + // The wait below already polls for the CDC data; the former fixed 30s + // pre-wait was folded into the budget (120s + 30s). given().ignoreExceptions() .await() - .atMost(120000, TimeUnit.MILLISECONDS) + .atMost(150000, TimeUnit.MILLISECONDS) .untilAsserted( () -> { Schema schema = loadIcebergSchema(); @@ -507,13 +508,11 @@ public class IcebergSinkCDCIT extends TestSuiteBase implements TestResource { // Init table data addTableColumn(MYSQL_DATABASE, SOURCE_TABLE, addField); insertAddColumnData(MYSQL_DATABASE, SOURCE_TABLE); - // Waiting 30s for source capture data - sleep(30000); - - // stream stage + // stream stage; the wait below already polls for the CDC data, the former + // fixed 30s pre-wait was folded into the budget (120s + 30s) given().ignoreExceptions() .await() - .atMost(120000, TimeUnit.MILLISECONDS) + .atMost(150000, TimeUnit.MILLISECONDS) .untilAsserted( () -> { Schema schema = loadIcebergSchema(); @@ -534,12 +533,11 @@ public class IcebergSinkCDCIT extends TestSuiteBase implements TestResource { String modifyField = "f_varchar"; modifyTableColumn(MYSQL_DATABASE, SOURCE_TABLE, modifyField, "text"); insertModifyColumnData(MYSQL_DATABASE, SOURCE_TABLE); - // Waiting 30s for source capture data - sleep(30000); - + // the wait below already polls for the CDC data, the former fixed 30s + // pre-wait was folded into the budget (120s + 30s) given().ignoreExceptions() .await() - .atMost(120000, TimeUnit.MILLISECONDS) + .atMost(150000, TimeUnit.MILLISECONDS) .untilAsserted( () -> { List<Record> records = loadIcebergTable(); @@ -557,12 +555,11 @@ public class IcebergSinkCDCIT extends TestSuiteBase implements TestResource { dropTableColumn(MYSQL_DATABASE, SOURCE_TABLE, addField); insertAfterDropColumnData(MYSQL_DATABASE, SOURCE_TABLE); - // Waiting 30s for source capture data - sleep(30000); - + // the wait below already polls for the CDC data, the former fixed 30s + // pre-wait was folded into the budget (120s + 30s) given().ignoreExceptions() .await() - .atMost(120000, TimeUnit.MILLISECONDS) + .atMost(150000, TimeUnit.MILLISECONDS) .untilAsserted( () -> { Schema schema = loadIcebergSchema(); @@ -648,12 +645,12 @@ public class IcebergSinkCDCIT extends TestSuiteBase implements TestResource { MYSQL_DATABASE, SOURCE_TABLE, newColumnName); executeSql(insertAfterRenameSql); - sleep(30000); // Wait for source capture data - - // Verify that column was renamed and data is correct + // Verify that column was renamed and data is correct. + // The wait below already polls for the CDC data; the former fixed 30s + // pre-wait was folded into the budget (120s + 30s). given().ignoreExceptions() .await() - .atMost(120000, TimeUnit.MILLISECONDS) + .atMost(150000, TimeUnit.MILLISECONDS) .untilAsserted( () -> { Schema schema = loadIcebergSchema(); @@ -692,13 +689,11 @@ public class IcebergSinkCDCIT extends TestSuiteBase implements TestResource { private void upsertAndCheckData(TestContainer container) throws InterruptedException, IOException { upsertDeleteSourceTable(MYSQL_DATABASE, SOURCE_TABLE); - // Waiting 30s for source capture data - sleep(30000); - - // stream stage + // stream stage; the wait below already polls for the CDC data, the former + // fixed 30s pre-wait was folded into the budget (120s + 30s) given().ignoreExceptions() .await() - .atMost(120000, TimeUnit.MILLISECONDS) + .atMost(150000, TimeUnit.MILLISECONDS) .untilAsserted( () -> { List<Record> records = loadIcebergTable(); @@ -717,13 +712,11 @@ public class IcebergSinkCDCIT extends TestSuiteBase implements TestResource { throws InterruptedException, IOException { // Init table data initSourceTableData(MYSQL_DATABASE, SOURCE_TABLE); - // Waiting 30s for source capture data - sleep(30000); - - // stream stage + // stream stage; the wait below already polls for the CDC data, the former + // fixed 30s pre-wait was folded into the budget (60s + 30s) given().ignoreExceptions() .await() - .atMost(60000, TimeUnit.MILLISECONDS) + .atMost(90000, TimeUnit.MILLISECONDS) .untilAsserted( () -> { Assertions.assertEquals(3, loadIcebergTable().size()); diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rabbitmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rabbitmq/RabbitmqIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rabbitmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rabbitmq/RabbitmqIT.java index 5ef10bdeed..3444edd01d 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rabbitmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rabbitmq/RabbitmqIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rabbitmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rabbitmq/RabbitmqIT.java @@ -51,6 +51,7 @@ import org.apache.seatunnel.format.json.JsonSerializationSchema; import org.apache.seatunnel.format.protobuf.ProtobufDeserializationSchema; import org.apache.seatunnel.format.protobuf.ProtobufSerializationSchema; +import org.awaitility.Awaitility; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeAll; @@ -420,7 +421,16 @@ public class RabbitmqIT extends TestSuiteBase implements TestResource { // send data to source queue before executeJob start in every testContainer initSourceData(sourceClient); - Thread.sleep(3000); + // Poll until the published messages are visible on the queue instead of a + // fixed sleep, so the job cannot start consuming before the data is there. + Awaitility.await() + .atMost(30, TimeUnit.SECONDS) + .pollInterval(1, TimeUnit.SECONDS) + .ignoreExceptions() + .until( + () -> + sourceClient.getChannel().messageCount(sourceQueueName) + >= TEST_DATASET.getValue().size()); // init consumer client before executeJob start in every testContainer RabbitmqClient sinkRabbitmqClient = getRabbitmqClient(sinkQueueName); @@ -532,10 +542,19 @@ public class RabbitmqIT extends TestSuiteBase implements TestResource { sendData(queue1, type1, 10); sendData(queue2, type2, 10); - // Wait briefly to ensure all messages are fully persisted and available in the RabbitMQ - // broker - // before the SeaTunnel job starts consuming. - Thread.sleep(5000); + // Poll until the published messages are visible on both queues instead of + // a fixed sleep, so the job cannot start consuming before the data is there. + try (RabbitmqClient probe1 = getRabbitmqClient(queue1); + RabbitmqClient probe2 = getRabbitmqClient(queue2)) { + Awaitility.await() + .atMost(30, TimeUnit.SECONDS) + .pollInterval(1, TimeUnit.SECONDS) + .ignoreExceptions() + .until( + () -> + probe1.getChannel().messageCount(queue1) >= 10 + && probe2.getChannel().messageCount(queue2) >= 10); + } // Execute the SeaTunnel synchronization job. // The job uses a multi-table configuration to consume from both queues simultaneously.
