He-Pin opened a new pull request, #3184:
URL: https://github.com/apache/pekko/pull/3184
### Motivation
Issue #3110 tracks stream operators that should honor `SupervisionStrategy`
for user functions.
`batchWeighted` already supervised `seed` and `aggregate`, but `costFn` in
`Batch.onPush()` still ran outside that boundary, so `costFn` exceptions failed
the stream unconditionally.
While adding `costFn` supervision, additional coupled exceptional-path
issues were found around pending-element handling in `flush()`/EOS handling
(state leakage and invalid pull-after-fail behavior). Those are fixed in the
same PR to keep supervision behavior consistent and safe.
### Modification
- Added supervision handling around `costFn(elem)` in `Batch.onPush()`:
- `Stop` → fail stage
- `Resume` → drop offending element and continue
- `Restart` → reset batch state and continue
- Hardened pending exceptional paths in `Batch`:
- clear `agg` immediately after push before processing `pending`
- evaluate `costFn(pending)` before `seed(pending)` in `flush()`
- guard follow-up pulls with `!isClosed(in)`
- complete stage at EOS when pending is dropped by `Resume`/`Restart`
after seed failure
- Expanded directional tests in `FlowBatchSpec`:
- explicit `Stop` and default `Stop`
- Resume vs Restart semantics
- pending/flush second `costFn` evaluation failures under
Resume/Restart/Stop
- EOS pending-seed failure under Restart
### Result
`batch` / `batchWeighted` now consistently apply supervision to `costFn`,
`seed`, and `aggregate`, and maintain correct state/pull behavior on
exceptional pending paths.
which is now Apache licensed
### Tests
- `sbt "stream-tests/Test/testOnly
org.apache.pekko.stream.scaladsl.FlowBatchSpec"` (14/14 passed)
- `sbt "stream/mimaReportBinaryIssues"` (clean)
### References
Refs #3110
--
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]
