He-Pin opened a new pull request, #3178: URL: https://github.com/apache/pekko/pull/3178
### Motivation Per the [stream error handling docs](https://pekko.apache.org/docs/pekko/current/stream/stream-error.html#supervision-strategies), operators that apply user-provided functions should consult the configured `SupervisionStrategy`. The `groupedWeighted` operator's `costFn` was called without a try-catch, so any exception failed the stream **unconditionally** — even when `Supervision.Resume` or `Supervision.Restart` was configured. This was inconsistent with `map`, `filter`, `batch`, etc. This is part of the meta-issue #3110 (add supervisor strategy support to stream operators that accept user functions). One operator per PR. ### Modification - Wrap `costFn(elem)` in `GroupedWeighted.onPush()` with a `try`/`catch` (`NonFatal`) that consults the `SupervisionStrategy` decider: - **Stop** → fail the stage (unchanged fail-fast behavior) - **Resume** → skip the offending element, **keep** the current partial group - **Restart** → **drop** the accumulated group and reset the weight counter - `decider` is a `lazy val`, so there is **zero overhead on the happy path** (it is only resolved when an exception actually occurs). - Add `SourceLocation.forLambda(costFn)` to `initialAttributes` for better failure attribution, matching the sibling `groupedWeightedWithin`. - Document supervision adherence in the Scala/Java DSL scaladoc and the `groupedWeighted` operator reference page (required by the supervision documentation policy in `stream-error.md`). ### Result `groupedWeighted` now adheres to the `SupervisionStrategy` attribute with well-defined group semantics under Resume/Restart, consistent with the rest of the stream operators. ### Tests - `sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowGroupedWeightedSpec"` — **14/14 passed**, including 5 new directional tests: - fails on `costFn` throw with `Supervision.Stop` - fails with no supervision attribute set (default Stop) — back-compat guard - `Resume` keeps the partially accumulated group and skips the failing element - `Resume` flushes the buffered group at completion when the last element throws - `Restart` drops the current group - `sbt "stream/mimaReportBinaryIssues"` — clean (binary compatible) ### References Refs #3110 This is a clean-room implementation written directly for Apache Pekko. -- 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]
