jerrypeng opened a new pull request, #59901:
URL: https://github.com/apache/spark/pull/59901
### What changes were proposed in this pull request?
Fail fast after a captured state-store write failure in
`StatefulStreamlineAggregationProcessor`:
- Share the existing final-flush exception check with per-input processing.
- Check at `process` entry and immediately after the cache lookup, before
merging or
returning an aggregate.
- Reject calls to the empty-global-aggregation entry point after a captured
failure.
- Add focused processor, completion-ordering, and HDFS recovery regression
tests for
aggregation state formats 1 and 2.
The first captured cause, exception wrapper, `NonFatal` boundary,
suppression of later
writes, and provider commit/abort ownership are preserved. There is no cache
redesign.
### Why are the changes needed?
Guava swallows exceptions from cache removal listeners. The processor
already captures
the first write failure and rethrows it at final flush, preventing partial
state from
committing. However, processing can continue and return intermediate
aggregates before
that flush.
For example, key A accumulates 5 + 10 = 15 in the cache. Its eviction write
fails before
persisting the value. If processing continues and A is reloaded for another
input of 20,
it can return 20 instead of 35. A failed final flush cannot retract results
already
observed downstream. The post-lookup check rejects the input whose eviction
fails before
it can merge or return an aggregate, stopping this continuation.
This completes the per-input failure handling of the streamline aggregation
operator
introduced in
[SPARK-58635](https://issues.apache.org/jira/browse/SPARK-58635)
([#57812](https://github.com/apache/spark/pull/57812)); final-flush
protection remains.
### Does this PR introduce _any_ user-facing change?
Yes. Affected streamline aggregation attempts fail earlier after a captured
write
failure, rather than potentially returning further aggregates before failing
at flush.
This affects RTM and the explicit internal streamline-aggregation
development option;
ordinary default microbatch aggregation is unchanged.
Healthy results and ordering, configurations, state schemas, and checkpoint
formats are
unchanged. No checkpoint migration is required. Rollback remains
format-compatible but
restores the previous failure behavior. This does not retract results
emitted before a
failure or add an end-to-end exactly-once sink guarantee.
### How was this patch tested?
SQL-core production and test sources compiled successfully with SBT 1.13.0,
Scala 2.13.18, and JDK 17.0.15. All 129 selected tests passed:
- `StatefulStreamlineAggregationProcessorSuite`: 18 tests.
- `StreamingAggregationSuite`, selecting `streamline aggregation` and
`switching between`: 35 tests.
- `RocksDBStateStoreStreamingAggregationSuite`, with the same filters: 70
tests,
covering snapshot and changelog checkpointing.
- `StreamlineStreamingAggregationRealTimeSuite`: all 6 tests.
```sh
build/sbt \
'sql/testOnly *StatefulStreamlineAggregationProcessorSuite' \
'sql/testOnly org.apache.spark.sql.streaming.StreamingAggregationSuite --
-z "streamline aggregation" -z "switching between"' \
'sql/testOnly
org.apache.spark.sql.streaming.RocksDBStateStoreStreamingAggregationSuite -- -z
"streamline aggregation" -z "switching between"' \
'sql/testOnly
org.apache.spark.sql.streaming.StreamlineStreamingAggregationRealTimeSuite' \
'sql/Compile/scalastyle' \
'sql/Test/scalastyle'
```
Both production and test Scala style checks passed with zero errors or
warnings.
As a negative control, compiled the unchanged production file from the base
revision
through a temporary SBT source override, retaining the new tests. All four
selected
eviction-failure cases failed because the expected immediate
`IllegalStateException`
was not thrown; no suite aborted. Restoring the fixed source made all 18
processor
tests pass again.
The new suite covers healthy eviction/reload, failures before and after
local mutation,
rejection of the triggering input, first-cause preservation, no further
state accesses,
flush-only failures, empty-global aggregation, flush-before-commit ordering,
and HDFS
abort/reopen/replay from committed state. The recovery test uses the real
HDFS provider,
but explicitly aborts/reopens it; it is not a full task/query retry test. No
new RocksDB
injected-failure or production performance result is claimed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
