You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "cshuo (via GitHub)" <gi...@apache.org> on 2026/10/15 02:28:03 UTC
[PR] fix(flink): serialize coordinator recovery with checkpoint commits [hudi]
cshuo opened a new pull request, #20339:
URL: https://github.com/apache/hudi/pull/20339
### Describe the issue this Pull Request addresses
Closes https://github.com/apache/hudi/issues/20331.
During same-graph global failover, `resetToCheckpoint()` can run recovery while the coordinator executor is still committing the same instant, producing duplicate completed timeline files and commit callbacks.
### Summary and Changelog
- Run the live coordinator's `restoreEvents()` through `executor.executeSync()`, ordering recovery after queued events and commits and finishing it before reset returns.
- Add a regression test that pauses a normal commit, triggers reset concurrently, and checks that reset waits and the instant is committed only once, with one physical completed file.
Validation: the new regression fails before the fix (including all three automatic retries). After the fix, the coordinator test class passes: 61 tests, 0 failures, 0 errors, 1 existing disabled test skipped. Checkstyle passes.
```bash
mvn -B -Dflink1.20 -pl hudi-flink-datasource/hudi-flink -am -Dtest=TestStreamWriteOperatorCoordinator -Dsurefire.failIfNoSpecifiedTests=false -DskipITs -DskipSparkTests -DskipScalaTests test
```
### Impact
Prevents overlapping checkpoint and recovery commits. Initial restoration during `start()` is unchanged; no public API or storage-format changes.
### Risk Level
Low. Recovery now waits for earlier executor work; a stalled commit can delay reset. Existing synchronous recovery already performs commit I/O before returning.
### Documentation Update
None.
### Contributor's checklist
- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Enough context is provided in the sections above
- [x] Adequate tests were added if applicable
π€ Generated with [Claude Code](https://claude.com/claude-code)
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): serialize coordinator recovery with checkpoint commits [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #20339:
URL: https://github.com/apache/hudi/pull/20339#discussion_r4279620989
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java:
##########
@@ -306,6 +311,74 @@ public void testCheckpointAndRestore(boolean isStreamingIndexWriteEnabled, boole
.filterCompletedInstants().containsInstant(instant));
}
+ @Test
+ void testGlobalRestoreWaitsForPendingCommit() throws Exception {
+ coordinator.close();
+ Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ conf.set(FlinkOptions.METADATA_ENABLED, false);
+ conf.set(FlinkOptions.TABLE_SERVICES_ENABLED, false);
+ coordinator = startCoordinator(conf, 2);
+ String instant = requestInstantTime(-1);
+ CompletableFuture<byte[]> checkpoint = new CompletableFuture<>();
+ coordinator.checkpointCoordinator(1, checkpoint);
+ coordinator.handleEventFromOperator(0, createOperatorEvent(0, instant, "par1", true, 0.1));
+ coordinator.handleEventFromOperator(1, createOperatorEvent(1, instant, "par2", true, 0.2));
+ byte[] checkpointData = checkpoint.get();
+
+ CountDownLatch commitEntered = new CountDownLatch(1);
+ CountDownLatch releaseCommit = new CountDownLatch(1);
+ AtomicInteger commitCalls = new AtomicInteger();
+ HoodieFlinkWriteClient client = Mockito.spy(coordinator.getWriteClient());
+ Mockito.doAnswer(invocation -> {
+ if (commitCalls.incrementAndGet() == 1) {
+ commitEntered.countDown();
+ assertTrue(releaseCommit.await(10, TimeUnit.SECONDS));
+ }
+ return invocation.callRealMethod();
+ }).when(client).commit(Mockito.eq(instant), Mockito.anyList(), Mockito.any(), Mockito.anyString(), Mockito.anyMap());
+ Field clientField = StreamWriteOperatorCoordinator.class.getDeclaredField("writeClient");
+ clientField.setAccessible(true);
+ clientField.set(coordinator, client);
+
+ MockOperatorCoordinatorContext ctx = (MockOperatorCoordinatorContext) coordinator.getContext();
Review Comment:
+1
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): serialize coordinator recovery with checkpoint commits [hudi]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #20339:
URL: https://github.com/apache/hudi/pull/20339#issuecomment-6183431013
## [Codecov](https://app.codecov.io/gh/apache/hudi/pull/20339?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 80.65%. Comparing base ([`b86a34b`](https://app.codecov.io/gh/apache/hudi/commit/b86a34b59e015b68796d88de8b18eea38c1739ae?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`66d578f`](https://app.codecov.io/gh/apache/hudi/commit/66d578f1473f578e5885f764b6af0a63d299d522?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 7 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #20339 +/- ##
============================================
- Coverage 80.71% 80.65% -0.06%
- Complexity 35134 35187 +53
============================================
Files 2553 2553
Lines 143524 143784 +260
Branches 17474 17511 +37
============================================
+ Hits 115839 115971 +132
- Misses 19743 19859 +116
- Partials 7942 7954 +12
```
| [Components](https://app.codecov.io/gh/apache/hudi/pull/20339/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Ξ | |
|---|---|---|
| [hudi-common](https://app.codecov.io/gh/apache/hudi/pull/20339/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.93% <ΓΈ> (-0.25%)` | :arrow_down: |
| [hudi-client](https://app.codecov.io/gh/apache/hudi/pull/20339/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.93% <ΓΈ> (+<0.01%)` | :arrow_up: |
| [hudi-flink](https://app.codecov.io/gh/apache/hudi/pull/20339/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `86.05% <ΓΈ> (-0.01%)` | :arrow_down: |
| [hudi-spark-datasource](https://app.codecov.io/gh/apache/hudi/pull/20339/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `74.18% <ΓΈ> (-0.01%)` | :arrow_down: |
| [hudi-utilities](https://app.codecov.io/gh/apache/hudi/pull/20339/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.20% <ΓΈ> (ΓΈ)` | |
| [hudi-cli](https://app.codecov.io/gh/apache/hudi/pull/20339/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.80% <ΓΈ> (ΓΈ)` | |
| [hudi-hadoop](https://app.codecov.io/gh/apache/hudi/pull/20339/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `71.17% <ΓΈ> (ΓΈ)` | |
| [hudi-sync](https://app.codecov.io/gh/apache/hudi/pull/20339/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `76.65% <ΓΈ> (ΓΈ)` | |
| [hudi-io](https://app.codecov.io/gh/apache/hudi/pull/20339/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.48% <ΓΈ> (-0.05%)` | :arrow_down: |
| [hudi-timeline-service](https://app.codecov.io/gh/apache/hudi/pull/20339/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.41% <ΓΈ> (ΓΈ)` | |
| [hudi-cloud](https://app.codecov.io/gh/apache/hudi/pull/20339/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.00% <ΓΈ> (ΓΈ)` | |
| [hudi-kafka-connect](https://app.codecov.io/gh/apache/hudi/pull/20339/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `54.48% <ΓΈ> (ΓΈ)` | |
| [Flag](https://app.codecov.io/gh/apache/hudi/pull/20339/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Ξ | |
|---|---|---|
| [hadoop-mr-java-client](https://app.codecov.io/gh/apache/hudi/pull/20339/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `43.94% <ΓΈ> (-0.01%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/hudi/pull/20339?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Ξ | |
|---|---|---|
| [...ache/hudi/sink/StreamWriteOperatorCoordinator.java](https://app.codecov.io/gh/apache/hudi/pull/20339?src=pr&el=tree&filepath=hudi-flink-datasource%2Fhudi-flink%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fhudi%2Fsink%2FStreamWriteOperatorCoordinator.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1mbGluay1kYXRhc291cmNlL2h1ZGktZmxpbmsvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvc2luay9TdHJlYW1Xcml0ZU9wZXJhdG9yQ29vcmRpbmF0b3IuamF2YQ==) | `83.15% <ΓΈ> (-0.07%)` | :arrow_down: |
... and [6 files with indirect coverage changes](https://app.codecov.io/gh/apache/hudi/pull/20339/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): serialize coordinator recovery with checkpoint commits [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
hudi-agent commented on code in PR #20339:
URL: https://github.com/apache/hudi/pull/20339#discussion_r4278488402
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java:
##########
@@ -306,6 +311,74 @@ public void testCheckpointAndRestore(boolean isStreamingIndexWriteEnabled, boole
.filterCompletedInstants().containsInstant(instant));
}
+ @Test
+ void testGlobalRestoreWaitsForPendingCommit() throws Exception {
+ coordinator.close();
+ Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ conf.set(FlinkOptions.METADATA_ENABLED, false);
+ conf.set(FlinkOptions.TABLE_SERVICES_ENABLED, false);
+ coordinator = startCoordinator(conf, 2);
+ String instant = requestInstantTime(-1);
+ CompletableFuture<byte[]> checkpoint = new CompletableFuture<>();
+ coordinator.checkpointCoordinator(1, checkpoint);
+ coordinator.handleEventFromOperator(0, createOperatorEvent(0, instant, "par1", true, 0.1));
+ coordinator.handleEventFromOperator(1, createOperatorEvent(1, instant, "par2", true, 0.2));
+ byte[] checkpointData = checkpoint.get();
+
+ CountDownLatch commitEntered = new CountDownLatch(1);
+ CountDownLatch releaseCommit = new CountDownLatch(1);
+ AtomicInteger commitCalls = new AtomicInteger();
+ HoodieFlinkWriteClient client = Mockito.spy(coordinator.getWriteClient());
+ Mockito.doAnswer(invocation -> {
+ if (commitCalls.incrementAndGet() == 1) {
+ commitEntered.countDown();
+ assertTrue(releaseCommit.await(10, TimeUnit.SECONDS));
+ }
+ return invocation.callRealMethod();
+ }).when(client).commit(Mockito.eq(instant), Mockito.anyList(), Mockito.any(), Mockito.anyString(), Mockito.anyMap());
+ Field clientField = StreamWriteOperatorCoordinator.class.getDeclaredField("writeClient");
+ clientField.setAccessible(true);
+ clientField.set(coordinator, client);
+
+ MockOperatorCoordinatorContext ctx = (MockOperatorCoordinatorContext) coordinator.getContext();
Review Comment:
π€ nit: could you split the reflection swap of `writeClient` into a small helper, or reuse an existing test hook if there is one? It would make the test body shorter and easier to follow.
<sub><i>β οΈ AI-generated; verify before applying. React π/π to flag quality.</i></sub>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): serialize coordinator recovery with checkpoint commits [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous closed pull request #20339: fix(flink): serialize coordinator recovery with checkpoint commits
URL: https://github.com/apache/hudi/pull/20339
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org