cshuo opened a new issue, #20331:
URL: https://github.com/apache/hudi/issues/20331
### Description
During a global failover within the same JobMaster/execution graph,
`StreamWriteOperatorCoordinator.resetToCheckpoint()` can recommit an instant
while the coordinator's event executor is still executing the normal
`notifyCheckpointComplete()` commit for that instant.
A deterministic coordinator-level reproduction produces **two physical
completed files with different completion times for the same requested
instant**, and invokes the configured commit callback twice. Both commit calls
succeed; the coordinator context reports no failure.
This concerns recovery of an **already-started coordinator instance**, not
initial restoration into a newly created coordinator before `start()`.
### Expected behavior
Recovery should be ordered after the coordinator's previously accepted
event/commit work. If the normal checkpoint-completion task has already
committed an instant, recovery should observe that result instead of starting a
second commit for the same instant.
### Trigger and code path
```text
Coordinator event executor JobManager main thread
-----------------------------------------
-----------------------------------
notifyCheckpointComplete(K) task
-> commitInstants(K)
-> writeClient.commit(T)
T has not reached COMPLETED yet
T's event buffer is still present
Same-graph global failover
Cancel old task executions
resetToCheckpoint(K)
-> restoreEvents(K)
-> T is not in completed
timeline
-> recommitInstant(T)
-> writeClient.commit(T)
again
Original commit resumes/completes
```
The normal commit must remain in progress long enough to overlap recovery;
stopping the TaskManager-side tasks does not drain the JobManager-side
coordinator executor.
At baseline commit
[`aabed46593960ff3553cdc1ace1a491987983702`](https://github.com/apache/hudi/commit/aabed46593960ff3553cdc1ace1a491987983702):
-
[`notifyCheckpointComplete()`](https://github.com/apache/hudi/blob/aabed46593960ff3553cdc1ace1a491987983702/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java#L318-L333)
executes commits on the event executor.
-
[`resetToCheckpoint()`](https://github.com/apache/hudi/blob/aabed46593960ff3553cdc1ace1a491987983702/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java#L336-L351)
calls `restoreEvents()` directly for a live coordinator, outside that executor.
-
[`recommitInstant()`](https://github.com/apache/hudi/blob/aabed46593960ff3553cdc1ace1a491987983702/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java#L570-L582)
decides to recommit from a completed-timeline snapshot. This check can run
before the original commit finishes.
-
[`doCommit()`](https://github.com/apache/hudi/blob/aabed46593960ff3553cdc1ace1a491987983702/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java#L682-L689)
removes the event buffer only after `writeClient.commit()` returns
successfully.
Flink 1.20.1's
[`DefaultScheduler.restartTasks()`](https://github.com/apache/flink/blob/release-1.20.1/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultScheduler.java#L396-L418)
restores state before redeploying tasks, but reuses the coordinator on this
path. Hudi's `executionAttemptFailed()` only clears the subtask gateway; it
does not stop/drain the event executor.
### Reproduction and results
The test uses the existing `TestStreamWriteOperatorCoordinator` fixtures, a
real single-thread `NonThrownExecutor`, and a spy around the real write client.
A latch pauses the first `commit()` invocation at entry. The test calls
`resetToCheckpoint()` on the same coordinator, then releases the first
invocation. The fixtures supply synthetic write-status metadata; the real
commit implementation writes actual timeline files to the local filesystem.
This test does not write and read back data rows. The callback is a local
Mockito mock.
| Observation | Normal commit completes before reset | Reset overlaps normal
commit |
| --- | ---: | ---: |
| `writeClient.commit()` calls | 1 | 2 |
| Physical completed files for the requested instant | 1 | 2 |
| Commit callback invocations | 1 | 2 |
| Coordinator failure | None | None |
Example files from the overlapping case:
```text
20261014165059320_20261014165059340.commit
20261014165059320_20261014165059354.commit
```
**Checking only the logical timeline count misses this.** Timeline v2
creates a new completion-time-qualified file on each completion.
[`TimelineLayout.filterHoodieInstantsByLatestState()`](https://github.com/apache/hudi/blob/aabed46593960ff3553cdc1ace1a491987983702/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineLayout.java#L111-L122)
groups by requested time/action and compares state, retaining the first
encountered entry when both states are COMPLETED. It does not resolve the
competing completion times by an explicit earliest/latest policy.
Confirmed effects are duplicate physical completion records and repeated
commit callbacks. Conflicting completion times also create ambiguity for
completion-time-based readers, but downstream missed/duplicate records have
**not** been reproduced in this test. This example does not demonstrate data
loss.
<details>
<summary>Test method to add to TestStreamWriteOperatorCoordinator at the
baseline commit</summary>
The method uses the existing class's coordinator setup, event-building, and
instant-request helpers. Additional types are fully qualified.
```java
@ParameterizedTest
@ValueSource(booleans = {false, true})
void testGlobalRestoreDoesNotCommitSameInstantTwice(boolean overlapCommit)
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));
CountDownLatch commitEntered = new CountDownLatch(1);
CountDownLatch releaseCommit = new CountDownLatch(1);
java.util.concurrent.atomic.AtomicInteger commitCalls = new
java.util.concurrent.atomic.AtomicInteger();
HoodieFlinkWriteClient client = Mockito.spy(coordinator.getWriteClient());
client.getConfig().setValue(org.apache.hudi.config.HoodieWriteCommitCallbackConfig.TURN_CALLBACK_ON,
"true");
org.apache.hudi.callback.HoodieWriteCommitCallback callback =
Mockito.mock(org.apache.hudi.callback.HoodieWriteCommitCallback.class);
Field callbackField =
org.apache.hudi.client.BaseHoodieClient.class.getDeclaredField("commitCallback");
callbackField.setAccessible(true);
callbackField.set(client, callback);
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();
NonThrownExecutor worker =
NonThrownExecutor.builder(Mockito.mock(Logger.class))
.exceptionHook((message, cause) -> ctx.failJob(new
HoodieException(message, cause)))
.waitForTasksFinish(true).build();
coordinator.setExecutor(worker);
try {
coordinator.notifyCheckpointComplete(1);
assertTrue(commitEntered.await(10, TimeUnit.SECONDS));
if (!overlapCommit) {
releaseCommit.countDown();
worker.close();
}
coordinator.resetToCheckpoint(1, checkpoint.get());
} finally {
releaseCommit.countDown();
worker.close();
}
HoodieTableMetaClient meta = StreamerUtil.createMetaClient(conf);
HoodieTimeline completed =
meta.reloadActiveTimeline().filterCompletedInstants();
List<String> physicalCommits =
meta.getStorage().listDirectEntries(meta.getTimelinePath()).stream()
.map(info -> info.getPath().getName())
.filter(name -> name.startsWith(instant + "_") &&
name.endsWith(".commit"))
.collect(java.util.stream.Collectors.toList());
int callbackCalls =
Mockito.mockingDetails(callback).getInvocations().size();
System.out.println("RESTORE_COMMIT overlap=" + overlapCommit + "
commitCalls=" + commitCalls.get()
+ " physicalCommits=" + physicalCommits + " callbackCalls=" +
callbackCalls
+ " completed=" +
completed.getInstantsAsStream().collect(java.util.stream.Collectors.toList())
+ " failure=" + ctx.getJobFailureReason());
assertNull(ctx.getJobFailureReason());
org.junit.jupiter.api.Assertions.assertAll(
() -> assertEquals(1, physicalCommits.size(), "One write must have one
physical completed file"),
() -> Mockito.verify(callback, Mockito.times(1)).call(Mockito.any()));
}
```
Run from the repository root after adding the method:
```bash
mvn -B -Dflink1.20 -pl hudi-flink-datasource/hudi-flink -am
'-Dtest=TestStreamWriteOperatorCoordinator#testGlobalRestoreDoesNotCommitSameInstantTwice'
-Dsurefire.failIfNoSpecifiedTests=false -DskipITs -DskipSparkTests
-DskipScalaTests test
```
Result: `Tests run: 2, Failures: 1, Errors: 0, Skipped: 0`. The normal-order
control passes; the overlapping case fails both the physical-file-count and
callback-count assertions. All three automatic reruns of the failing case
reproduce both effects.
This is a component-level reproduction with real timeline writes, not a
Flink MiniCluster test. The latch models an in-progress commit surviving task
cancellation/restart delay. It does not measure how frequently this
interleaving occurs in a running cluster. After adding synchronous restore
ordering, the test's latch arrangement would need to release the original
commit from another thread while reset is waiting.
</details>
### Environment
- Hudi: `1.3.0-SNAPSHOT`, commit `aabed46593960ff3553cdc1ace1a491987983702`.
- Flink: `1.20.1`, Maven profile `-Dflink1.20`.
- JDK: Eclipse Temurin `11.0.27`; Maven `3.9.9`.
- Data table: default COPY_ON_WRITE/upsert, table version 10, timeline
layout v2.
- The reproduction disables metadata writes and table services to isolate
data-table commit behavior, and enables the commit callback with a local mock.
### Suggested fix direction
Order the live-coordinator `resetToCheckpoint()` recovery on the same event
executor as normal commits, waiting for it to complete before returning, for
example using `executor.executeSync(() -> restoreEvents(checkpointID), ...)` in
that branch. This lets the normal commit finish and remove its buffer before
recovery inspects pending work. The initial restore path before `start()` has
no old executor work and is a different lifecycle case.
The implementation should consider Flink's reset ordering and JobManager
main-thread blocking constraints. The existing synchronous `restoreEvents()`
already performs commit I/O on the caller thread; using the executor also
introduces waiting for earlier queued work.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
