You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "wombatu-kun (via GitHub)" <gi...@apache.org> on 2026/05/24 04:21:34 UTC
[PR] Flink: Fix flaky TestMonitorSource.testStateRestore [iceberg]
wombatu-kun opened a new pull request, #16548:
URL: https://github.com/apache/iceberg/pull/16548
Closes #16546.
## Summary
`TestMonitorSource.testStateRestore` (the `testStateRestore(File, ClusterClient)` variant in the Flink v2.0/v2.1 trees) intermittently fails with a `TimeoutException` from `CollectingSink.poll` ([example CI run](https://github.com/apache/iceberg/actions/runs/26314549114/job/77471777704)). The timeout only means the sink queue stayed empty for 5s; the real cause is a savepoint-completion race in the shared test helper, not slow startup.
`OperatorTestBase.closeJobClient(JobClient, File)` discarded the `CompletableFuture<String>` returned by `stopWithSavepoint` and instead waited for the savepoint *directory* to appear on disk. That directory is created early in the savepoint process, before the `_metadata`/state files finish writing. Phase 2 of the test then restores from that path via `clusterClient.submitJob`; when the restore races savepoint completion, the restored job never comes up and emits nothing, so the poll times out.
## What changed
- `OperatorTestBase.closeJobClient` now awaits the `stopWithSavepoint(...)` future and returns the path it resolves to, so the savepoint is guaranteed to be fully written before any job restores from it. This mirrors the existing idiom in `TestIcebergSourceFailover.testBoundedWithSavepoint`, which awaits the savepoint future with `.get()`. The only caller that passes a non-null savepoint directory is `TestMonitorSource.testStateRestore`, so the change is scoped to this test.
- As a backstop for restored-job startup latency on busy CI, the Phase 2 poll in `testStateRestore` is raised from 5s to 30s. The assertion stays strict — a genuine re-read emits a non-empty event quickly and still fails fast — so the longer timeout only extends the wait for the (correct) first event, mirroring the "deterministic fix + generous timeout backstop" pattern used when `TestIcebergSourceFailover` was de-flaked.
Both changes are applied identically to the Flink v2.0 and v2.1 trees. The v1.20 variant uses `env.executeAsync` rather than `clusterClient.submitJob` and is not affected.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Fix flaky TestMonitorSource.testStateRestore [iceberg]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #16548:
URL: https://github.com/apache/iceberg/pull/16548#discussion_r3308859605
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestMonitorSource.java:
##########
@@ -221,7 +221,8 @@ void testStateRestore(@TempDir File savepointDir) throws Exception {
try {
clientWithSavepoint = env.executeAsync("Table Change Source test with savepoint");
- assertThat(resultWithSavepoint.poll(Duration.ofSeconds(5L))).isEqualTo(EMPTY_EVENT);
+ // Restoring from a savepoint on a busy cluster may take longer than the default 5s poll
+ assertThat(resultWithSavepoint.poll(Duration.ofSeconds(30L))).isEqualTo(EMPTY_EVENT);
Review Comment:
@pvary Ok, i reverted timeouts to 5s
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Fix flaky TestMonitorSource.testStateRestore [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #16548:
URL: https://github.com/apache/iceberg/pull/16548#issuecomment-4532062447
Is this generated by the help of an AI agent?
If so, please check https://iceberg.apache.org/contribute/#guidelines-for-ai-assisted-contributions
The stopWithSavepoint().get() seems like a good solution, but I think it should be changed in all Flink versions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Fix flaky TestMonitorSource.testStateRestore [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary merged PR #16548:
URL: https://github.com/apache/iceberg/pull/16548
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Fix flaky TestMonitorSource.testStateRestore [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #16548:
URL: https://github.com/apache/iceberg/pull/16548#discussion_r3303409990
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestMonitorSource.java:
##########
@@ -221,7 +221,8 @@ void testStateRestore(@TempDir File savepointDir) throws Exception {
try {
clientWithSavepoint = env.executeAsync("Table Change Source test with savepoint");
- assertThat(resultWithSavepoint.poll(Duration.ofSeconds(5L))).isEqualTo(EMPTY_EVENT);
+ // Restoring from a savepoint on a busy cluster may take longer than the default 5s poll
+ assertThat(resultWithSavepoint.poll(Duration.ofSeconds(30L))).isEqualTo(EMPTY_EVENT);
Review Comment:
Are we sure we need this? Wasn't this only a concurrency issue?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Fix flaky TestMonitorSource.testStateRestore [iceberg]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #16548:
URL: https://github.com/apache/iceberg/pull/16548#discussion_r3308748293
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestMonitorSource.java:
##########
@@ -221,7 +221,8 @@ void testStateRestore(@TempDir File savepointDir) throws Exception {
try {
clientWithSavepoint = env.executeAsync("Table Change Source test with savepoint");
- assertThat(resultWithSavepoint.poll(Duration.ofSeconds(5L))).isEqualTo(EMPTY_EVENT);
+ // Restoring from a savepoint on a busy cluster may take longer than the default 5s poll
+ assertThat(resultWithSavepoint.poll(Duration.ofSeconds(30L))).isEqualTo(EMPTY_EVENT);
Review Comment:
nope
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Fix flaky TestMonitorSource.testStateRestore [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #16548:
URL: https://github.com/apache/iceberg/pull/16548#discussion_r3308678493
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestMonitorSource.java:
##########
@@ -221,7 +221,8 @@ void testStateRestore(@TempDir File savepointDir) throws Exception {
try {
clientWithSavepoint = env.executeAsync("Table Change Source test with savepoint");
- assertThat(resultWithSavepoint.poll(Duration.ofSeconds(5L))).isEqualTo(EMPTY_EVENT);
+ // Restoring from a savepoint on a busy cluster may take longer than the default 5s poll
+ assertThat(resultWithSavepoint.poll(Duration.ofSeconds(30L))).isEqualTo(EMPTY_EVENT);
Review Comment:
Have you actually seen a case where a timeout was reached even with your fix already in place?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Fix flaky TestMonitorSource.testStateRestore [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #16548:
URL: https://github.com/apache/iceberg/pull/16548#issuecomment-4555390472
Merged to main.
Thanks @wombatu-kun for the fix!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Fix flaky TestMonitorSource.testStateRestore [iceberg]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #16548:
URL: https://github.com/apache/iceberg/pull/16548#discussion_r3303841651
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestMonitorSource.java:
##########
@@ -221,7 +221,8 @@ void testStateRestore(@TempDir File savepointDir) throws Exception {
try {
clientWithSavepoint = env.executeAsync("Table Change Source test with savepoint");
- assertThat(resultWithSavepoint.poll(Duration.ofSeconds(5L))).isEqualTo(EMPTY_EVENT);
+ // Restoring from a savepoint on a busy cluster may take longer than the default 5s poll
+ assertThat(resultWithSavepoint.poll(Duration.ofSeconds(30L))).isEqualTo(EMPTY_EVENT);
Review Comment:
The deterministic fix in `closeJobClient` removes the savepoint-completion race, which was the actual cause of the failure here. I kept the timeout bump as a separate, defensive measure: restoring a job on a busy CI runner (resubmit → schedule → deploy tasks) can occasionally take longer than 5s even when the savepoint is fully written, so the larger timeout guards against that kind of CI lag.
It doesn't slow the normal run down, though — `poll(...)` returns as soon as the expected event arrives, so on an unloaded runner / in the happy path the extra time is never actually spent; the 30s only caps how long we're willing to wait before failing. That's why I'd lean towards keeping it as a low-cost robustness backstop.
That said, I don't feel strongly about it — if you'd rather rely solely on the deterministic fix and keep the change minimal, just let me know and I'll revert the timeout part.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Fix flaky TestMonitorSource.testStateRestore [iceberg]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on PR #16548:
URL: https://github.com/apache/iceberg/pull/16548#issuecomment-4532508098
@pvary thanks for the review!
On the AI question: yes, this PR was prepared with AI assistance (drafting the fix, the refactor, and the test changes). Per the AI-assisted contribution guidelines I reviewed and verified it end-to-end — I understand the savepoint-completion race and ran the affected tests locally — and kept the wording and style aligned with the codebase. Happy to walk through any part during review.
On applying it to all Flink versions: done in 806876ff1. I'd originally scoped it to v2.0/v2.1 because the observed CI failure was the `testStateRestore(File, ClusterClient)` variant that only exists there, but you're right that the same latent race lives in v1.20's `closeJobClient`. PR description is updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org