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-13067-0959c444ff7ab74dadea079f94fb91f6afff077e in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 9544fb8cf21843239bf835ca1ac8f82afe0ddd96 Author: Jast <[email protected]> AuthorDate: Sat Oct 31 12:14:14 2026 +0000 [Test][Connector-V2][Iceberg] Poll for CDC capture and checkpoint commits instead of fixed sleeps (#13067) Co-authored-by: jast <[email protected]> --- .../e2e/connector/iceberg/IcebergSinkCDCIT.java | 36 +++++++++++-- .../iceberg/IcebergSinkParallelCommitIT.java | 60 +++++++++++++++++----- 2 files changed, 81 insertions(+), 15 deletions(-) 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 0099c3f586..79b6a168af 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 @@ -69,7 +69,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.connectors.seatunnel.iceberg.config.IcebergCatalogType.HADOOP; import static org.awaitility.Awaitility.given; @@ -326,7 +325,18 @@ public class IcebergSinkCDCIT extends TestSuiteBase implements TestResource { MYSQL_DATABASE, SOURCE_TABLE, modifyTypeField1, modifyTypeField2); executeSql(insertTypeColumnsSql); - sleep(30000); // Wait for source capture data + // Wait until the CDC source has captured the pre-ALTER record (it becomes visible in + // the sink table) instead of waiting a fixed time, so the schema change is applied + // only after the old-schema row was consumed. + given().ignoreExceptions() + .await() + .atMost(120000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> + Assertions.assertTrue( + icebergTableHasRecordWithId(300), + "Record 300 should be captured before the column type" + + " change")); // Now modify multiple column types in a single ALTER TABLE statement String modifyTypesSql = @@ -428,7 +438,18 @@ public class IcebergSinkCDCIT extends TestSuiteBase implements TestResource { MYSQL_DATABASE, SOURCE_TABLE, modifyField1, modifyField2); executeSql(insertModifyColumnsSql); - sleep(30000); // Wait for source capture data + // Wait until the CDC source has captured the pre-ALTER record (it becomes visible in + // the sink table) instead of waiting a fixed time, so the schema change is applied + // only after the old-schema row was consumed. + given().ignoreExceptions() + .await() + .atMost(120000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> + Assertions.assertTrue( + icebergTableHasRecordWithId(400), + "Record 400 should be captured before the column type" + + " change")); // Now modify multiple columns in a single ALTER TABLE statement String modifyColumnsSql = @@ -748,6 +769,15 @@ public class IcebergSinkCDCIT extends TestSuiteBase implements TestResource { return results; } + private boolean icebergTableHasRecordWithId(int id) { + for (Record record : loadIcebergTable()) { + if (Integer.valueOf(id).equals(record.getField("id"))) { + return true; + } + } + return false; + } + @NotNull private static IcebergTableLoader getTableLoader() { Map<String, Object> configs = new HashMap<>(); Map<String, Object> catalogProps = new HashMap<>(); diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iceberg/IcebergSinkParallelCommitIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iceberg/IcebergSinkParallelCommitIT.java index 0698a83964..617a631b2f 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iceberg/IcebergSinkParallelCommitIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iceberg/IcebergSinkParallelCommitIT.java @@ -244,12 +244,15 @@ public class IcebergSinkParallelCommitIT extends TestSuiteBase { * <ol> * <li>Start a streaming job with {@code parallelism=2}. {@code split.read-interval} keeps * FakeSource alive long enough to trigger several checkpoints. - * <li>Wait 9 s — at least 3 checkpoints complete and their snapshots land in Iceberg. + * <li>Wait until at least one checkpoint has been committed and its snapshots have landed in + * Iceberg — bounded polling on committed state, not a fixed sleep. * <li>Savepoint the job (stops it and persists the last committed state). * <li>Record the set of {@code seatunnel.checkpoint-id} values already present in the table. * <li>Restore the job from the savepoint. {@code IcebergAggregatedCommitter.restoreCommit()} * must detect the already-committed checkpoint via the snapshot summary and skip it. - * <li>Let the restored job run and produce additional checkpoints, then verify: + * <li>Wait until the restored job commits a checkpoint-id that was absent before the + * savepoint — this proves the restored run is live before any uniqueness claims — then + * verify: * <ul> * <li>Every snapshot in the table has a <em>unique</em> checkpoint-id — no checkpoint * was committed more than once. @@ -278,8 +281,18 @@ public class IcebergSinkParallelCommitIT extends TestSuiteBase { } }); - // checkpoint.interval=3s × 3 = 9 s - Thread.sleep(9_000); + // Wait until at least one checkpoint has been committed to the table instead of + // waiting a fixed number of checkpoint intervals, so the savepoint always happens + // after real commits even when checkpoints run slower under CI load. + given().ignoreExceptions() + .await() + .atMost(120000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> + Assertions.assertFalse( + collectCheckpointIds(loadRecoveryTable()).isEmpty(), + "At least one checkpoint must have been committed" + + " before the savepoint")); Assertions.assertEquals( 0, @@ -294,14 +307,37 @@ public class IcebergSinkParallelCommitIT extends TestSuiteBase { "At least one checkpoint must have been committed before the savepoint"); log.info("Checkpoint-ids committed before savepoint: {}", checkpointIdsBefore); - CompletableFuture.supplyAsync( - () -> { - try { - return container.restoreJob(recoveryConf, String.valueOf(jobId)); - } catch (Exception e) { - throw new RuntimeException(e); - } - }); + CompletableFuture<Container.ExecResult> restoreRun = + CompletableFuture.supplyAsync( + () -> { + try { + return container.restoreJob(recoveryConf, String.valueOf(jobId)); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + + // The pre-savepoint snapshot set already satisfies every uniqueness and occurrence + // assertion below, so those checks alone cannot prove the restored run did anything. + // Wait until the restored job commits a checkpoint-id that was absent before the + // savepoint before making any recovery claims. The restored job is streaming, so the + // restore command only returns once the job stops — it must never be joined here; + // only a non-blocking fail-fast on an already-failed restore is done. + given().ignoreExceptions() + .await() + .atMost(120000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> { + Assertions.assertFalse( + restoreRun.isDone() && restoreRun.isCompletedExceptionally(), + "restoreJob must not fail while waiting for its commits"); + Set<String> current = collectCheckpointIds(loadRecoveryTable()); + current.removeAll(checkpointIdsBefore); + Assertions.assertFalse( + current.isEmpty(), + "The restored job must commit at least one new checkpoint-id" + + " before recovery can be verified"); + }); // Wait long enough for the restored job to finish (4 splits × 3 s + buffer) given().ignoreExceptions()
