nsivabalan opened a new pull request, #20429: URL: https://github.com/apache/hudi/pull/20429
> **WIP** — opened as a draft for early alignment; open items are listed at the end. ### Describe the issue this Pull Request addresses When `HoodieStreamer` re-creates its write client (e.g. on a schema change), `StreamSync.reInitWriteClient` closes the previous client and the async compaction/clustering services are switched to the new one afterwards. A compaction or clustering already running on the previous client keeps using it after it was closed. If that run fails: - the async service stops with an error (the service loop has no per-instant error handling); - the continuous ingestion loop only checks the service when it next schedules a compaction/clustering, so ingestion keeps committing without table services until then, and then fails with `Async compaction failed` but without the cause; - the job exits and has to be restarted, at which point the pending instant is re-run. The Spark streaming sink shares the same async services and the same exposure (discussed on #20415, see #20400); it will be handled separately on top of this. ### Summary and Changelog - `AsyncCompactService` / `AsyncClusteringService`: each run records the write client it started with. A run that fails after the service was switched to a new write client is treated as retryable: the instant is enqueued again and runs with the new client (the failed attempt is rolled back by `compact`/`cluster`), and the service keeps running. Any other failure stops the service as before. Retries are bounded: a retried run that fails again without another swap is a regular failure. - `HoodieAsyncService`: records the failure the service stopped with (`getFailure()`); `hasError` is now `volatile` since callers poll it from other threads. - `HoodieStreamer` (continuous mode): checks both async table services at the start of every ingestion round and stops with the service failure as the cause. - `StreamSync.reInitWriteClient`: switches the async services to the new write client before closing the previous one, so a run that fails on the closed client is recognised as retryable. Tests: - `TestAsyncCompactionWriteClientSwap` (real write clients, MOR table): a compaction is held mid-run, the service is switched to a new client and the previous one closed, and the run fails — the instant is rolled back, re-run with the new client and committed, with the service still running. Without a swap, the service stops with the failure recorded and the compaction left pending. - `TestHoodieStreamerAsyncTableServiceFailure`: a continuous MOR `HoodieStreamer` whose async compaction fails (payload that throws only in compaction tasks) stops with that failure as the cause. ### Impact Writers that re-create their write client while async table services run (HoodieStreamer on schema changes) no longer lose the job when an in-flight compaction/clustering fails because of the swap. Non-recoverable async table service failures stop `HoodieStreamer` on the next ingestion round with the actual cause. ### Risk Level medium. Changes failure handling of the shared async compaction/clustering services and the write client swap order in `StreamSync`. Verified with the new tests plus existing `TestAsyncCompaction` (11/11) and continuous-mode `TestHoodieDeltaStreamer` tests (COW/MOR upserts, async clustering; 10/10). ### Documentation Update none ### Open items (WIP) - [ ] Ingestion notices a failed service at the start of its next round; with a long `--min-sync-interval-seconds` it could interrupt the sleep instead. - [ ] `StreamSync.reInitWriteClient`: close the previous client in a `finally` in case `onInitializingHoodieWriteClient` throws. - [ ] Functional test for the clustering retry (same code path as compaction). - [ ] Streamer-level test with a real schema change landing during a compaction. - [ ] Apply the same handling to the Spark streaming sink (#20415). ### Contributor's checklist - [x] 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
