fhan688 opened a new pull request, #19960: URL: https://github.com/apache/hudi/pull/19960
### Describe the issue this Pull Request addresses Closes https://github.com/apache/hudi/issues/19902 The streaming write task obtains its instant time from the coordinator over a Flink coordination RPC: `Correspondent.requestInstantTime` -> `StreamWriteOperatorCoordinator.handleInstantRequest`. Today that handler runs synchronously on the RPC path: it waits for prior-checkpoint instants to complete and then acquires the table lock inside `startInstant()` (reload timeline, `preTxn`, `startCommit`, transition to inflight). When the table lock is held by another actor (for example an async cleaning / table service), the handler blocks past the coordination RPC ask timeout (`pekko.ask.timeout`, default 10s). The ask then fails and the write task treats it as a coordinator failure, triggering a spurious pipeline restart even though nothing is actually wrong — the instant would have been created a moment later. ### Summary and Changelog Make the instant-time request an asynchronous request-reply protocol so the RPC always returns in O(1), independent of how long instant creation takes. - **Coordinator (`StreamWriteOperatorCoordinator`)**: `handleInstantRequest` no longer blocks. It reports the current status for the checkpoint — `READY` / `PENDING` /`FAILED` — and hands the blocking creation off to the existing single-thread `instantRequestExecutor`. Exactly one creation is submitted per checkpoint via `instantOps.computeIfAbsent(...)` + an `AtomicBoolean` claim, so concurrent writers for the same checkpoint share one instant. - **Idempotency**: the checkpoint -> instant mapping in `EventBuffers` is authoritative, so a lost `READY` reply is recovered by the next poll without creating a second instant. - **Generation fencing**: a new `volatile long epoch` is bumped on `start()`, `resetToCheckpoint()` and `close()`. An in-flight creation carries the epoch it was submitted under and refuses to publish its result once the generation moves on, so a failover/close never installs a stale instant. `isClosing` rejects new requests during shutdown, and `waitForTasksFinish(true)` drains a running creation before the write client is closed. - **Client (`Correspondent`)**: `requestInstantTime` polls with capped exponential backoff plus full jitter under a single deadline. `PENDING` keeps polling without resetting the deadline, `FAILED` fails fast, and transient transport errors retry within the same budget. The poll budget is `write.commit.ack.timeout` (`FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT`, default 300000ms). - **Signature**: `requestInstantTime(checkpointId)` becomes `requestInstantTime(checkpointId, pollBudgetMs)`; callers in `AbstractStreamWriteFunction` and `BulkInsertWriteFunction` pass `WRITE_COMMIT_ACK_TIMEOUT`. Test mocks/wrappers were updated accordingly. ### Impact No new user-facing config and no end-user public API change — `Correspondent` and the coordinator request handling are internal. Behavioral impact: a streaming write pipeline no longer restarts spuriously when a table-lock holder (e.g. cleaning) outlasts the coordination RPC ask timeout. The overall time a writer is willing to wait for an instant is now bounded by `write.commit.ack.timeout` rather than by `pekko.ask.timeout`. ### Risk Level medium This touches the coordinator instant-time path and its failover/close/reset handling. Verification: added a focused integration test(`TestStreamWriteOperatorCoordinator#testInstantRequestPollsWhileCreationBlockedThenSucceeds`) that holds the lock longer than the RPC ask timeout but shorter than the poll budget and asserts a prompt `PENDING` reply, no pipeline restart, eventual success, and exactly one instant across concurrent writers; existing `hudi-flink` sink coordinator tests continue to cover the commit / recommit / failover paths. ### Documentation Update none ### Contributor's checklist - [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute) - [ ] Enough context is provided in the sections above - [ ] Adequate tests were added if applicable -- 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]
