This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 9544fb8cf2 [Test][Connector-V2][Iceberg] Poll for CDC capture and
checkpoint commits instead of fixed sleeps (#13067)
9544fb8cf2 is described below
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()
