nsivabalan opened a new pull request, #20445:
URL: https://github.com/apache/hudi/pull/20445
### Describe the issue this Pull Request addresses
Stacked on #20444 (review only the last commit).
Since #20415, the Spark streaming sink re-creates its write client when the
writer schema changes, hands the new client to its async compaction/clustering
services and closes the previous one. With #20444 the switch does not wait for
a running compaction/clustering (which can take an hour), so a run that is
still using the previous client can fail once that client is closed. Today such
a failure stops the async service, and the next micro-batch fails the streaming
query with `IllegalStateException("Async Compactor shutdown unexpectedly")`,
without the cause — although the failure was caused by the sink itself and the
run would succeed with the new client.
HoodieStreamer is left as it is (1.2.0 behaviour); its handling of the same
case is discussed in #20429.
### Summary and Changelog
Failures of the sink's async table service runs are now classified as
retryable or not:
- **Retryable**: the run failed while using a write client the sink has
retired (closed when it re-created its client). The service waits until it has
been switched to the new client, then runs the instant again if it is still
pending (the run rolls back the failed inflight instant first, as on any
re-run). The query keeps running.
- **Not retryable**: any other failure. As before, the service stops and the
next micro-batch fails the query, now with the service's failure as the cause.
Changes:
- `AsyncCompactService` / `AsyncClusteringService`: per-run failure handling
with an opt-in hook, `isRetryableFailure(runClient, failure)`, `false` by
default, so services that do not override it (HoodieStreamer's) behave as
before. A retry waits for `updateWriteClient` to switch away from the failed
run's client and checks that the instant is still pending.
- `SparkStreamingAsyncCompactService` /
`SparkStreamingAsyncClusteringService`: take a predicate telling whether a
client was retired by the sink, and retry failures of runs that used one.
- `HoodieStreamingSink`: marks the previous client as retired, closes it,
then switches the services to the new client (weak set, so retired clients are
not retained); `addBatch` fails with the service's failure as the cause.
- `HoodieAsyncService`: records the failure the service stopped with
(`getFailure()`); `hasError` and `shutdownRequested` are volatile.
### Impact
A streaming query whose writer schema changes while an async compaction or
clustering is running no longer fails when that run loses its client: the run
is retried with the new client. Ingestion never waits for the running table
service. Non-retryable table service failures still fail the query, with the
cause attached. HoodieStreamer is unchanged.
Tests:
- `TestAsyncTableServiceRetry` (compaction and clustering): a retryable
failure waits for the new client and re-runs the instant with it; an instant
that is no longer pending is not re-run; by default a failure is not retried
and the service stops with it as `getFailure()`.
-
`TestStructuredStreaming#testStructuredStreamingRetriesAsyncCompactionFailedByWriteClientSwap`:
holds an async compaction, evolves the table schema so that the sink
re-creates its write client, asserts that micro-batch completes while the
compaction is still held, then fails the held compaction and checks that it is
retried and committed and that the query keeps running.
-
`TestStructuredStreaming#testStructuredStreamingFailsWithCauseWhenAsyncCompactionFails`:
a compaction failure unrelated to a client swap fails the query, with that
failure in the cause chain.
### Risk Level
medium. Changes failure handling of the async table services, behind a hook
that only the streaming sink's services enable.
### Documentation Update
None.
### 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]
