He-Pin opened a new pull request, #3180: URL: https://github.com/apache/pekko/pull/3180
### 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 `groupedWeightedWithin` operator's `costFn` was called without a try-catch, so any exception failed the stream **unconditionally** — even under `Supervision.Resume`/`Restart`. 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)` inside `GroupedWeightedWithin.nextElement()` with a `try`/`catch` (`NonFatal`) that consults the `SupervisionStrategy` decider. The cost is computed **before** any state mutation, so the happy path is unchanged: - **Stop** → fail the stage (unchanged fail-fast behavior) - **Resume** → skip the offending element, **keep** the current group (weight/number accounting is preserved) - **Restart** → **drop** the in-progress group via a new `resetGroupState()` helper, and **reschedule** the time-window timer so the fresh group gets a full window - `decider` is a `lazy val` → zero overhead on the happy path. - Document supervision adherence on **both** `groupedWeightedWithin` overloads in the Scala/Java DSL scaladoc and the operator reference page. ### Result `groupedWeightedWithin` now adheres to the `SupervisionStrategy` attribute with well-defined Resume/Restart semantics. ### Tests - `sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowGroupedWithinSpec"` — **21/21 passed**, including 5 new deterministic directional tests (Stop, default Stop, Resume keeps group, Resume preserves weight accounting, Restart drops group). They use a 30s window that never fires, so they are deterministic and run in PR validation (intentionally not tagged `TimingTest`). - `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]
