This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 5b80b55b702e646391557ad11f283bf2115eab53 Author: Vova Kolmakov <[email protected]> AuthorDate: Thu Sep 17 12:18:00 2026 +0700 [test] Wait for a commit past the aborted savepoint after region failover (#9907) --- .../AppendTableSavepointTagFailoverITCase.java | 33 +++++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableSavepointTagFailoverITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableSavepointTagFailoverITCase.java index 49198b4f51..0ca7387326 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableSavepointTagFailoverITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableSavepointTagFailoverITCase.java @@ -250,10 +250,18 @@ public class AppendTableSavepointTagFailoverITCase extends AbstractTestBase { ExceptionUtils.findThrowable( e, CheckpointException.class)) .isPresent()); + // Also guards the wait below: with no injected failure there is no savepoint id. + assertThat(FailOnSavepointOperator.hasFailed()) + .describedAs("the injected failure never fired, so no region failover happened") + .isTrue(); assertThat(savepointTags(table)).isEmpty(); // Wait until the job has recovered and resumed committing after the region failover. - waitUntilRecoveredAndCommitting(table); + // "A new snapshot appeared" is not that: the coordinator commits asynchronously, so a + // commit already in flight when the region failed can bump the snapshot id while the + // region is still restarting, and the savepoint below would then be rejected because + // not all tasks are running. + waitUntilCommittedPast(table, FailOnSavepointOperator.savepointCheckpointId()); // Savepoint 2: after recovery it must be tagged correctly, proving the region failover // did not break the coordinator's auto-tag state. @@ -481,6 +489,25 @@ public class AppendTableSavepointTagFailoverITCase extends AbstractTestBase { throw new IllegalStateException("job did not resume committing after region failover"); } + /** + * Waits until a snapshot commits with an identifier past {@code checkpointId}. Commit + * identifiers are Flink checkpoint ids, and a checkpoint only completes once every task has + * acknowledged it, so a commit past the aborted savepoint proves the failed region is running + * again - which is what triggering another savepoint requires. + */ + private void waitUntilCommittedPast(FileStoreTable table, long checkpointId) throws Exception { + long deadline = System.currentTimeMillis() + WAIT_TIMEOUT_MILLIS; + while (System.currentTimeMillis() < deadline) { + Snapshot latest = table.snapshotManager().latestSnapshot(); + if (latest != null && latest.commitIdentifier() > checkpointId) { + return; + } + Thread.sleep(200); + } + throw new IllegalStateException( + "no checkpoint completed past the aborted savepoint within timeout"); + } + /** * Waits until at least {@code count} more snapshots commit past the current latest. * Restore-time tag work runs before the resumed job commits again, so requiring several fresh @@ -581,6 +608,10 @@ public class AppendTableSavepointTagFailoverITCase extends AbstractTestBase { return savepointCheckpointId; } + static boolean hasFailed() { + return FAILED.get(); + } + @Override public void processElement(StreamRecord<RowData> element) { output.collect(element);
