[PR] feat: add supervision strategy support for groupedAdjacentBy/Weighted [pekko]

["He-Pin (via GitHub)" <gi...@apache.org>]

He-Pin opened a new pull request, #3181:
URL: https://github.com/apache/pekko/pull/3181

   ### 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 `groupedAdjacentBy` and `groupedAdjacentByWeighted` operators apply **two** user functions — a key function `f` and (for the weighted variant) a `costFn` — but both were 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 (group) per PR.
   
   ### Modification
   
   - Wrap **both** the `costFn` and the key function calls in `GroupedAdjacentByWeighted` with a `try`/`catch` (`NonFatal`) that consults the `SupervisionStrategy` decider via a shared `superviseUserFn` helper:
     - **Stop** → fail the stage (unchanged fail-fast behavior)
     - **Resume** → skip the offending element, **keep** the current group
     - **Restart** → **drop** the in-progress group (an already-completed group awaiting downstream demand is preserved)
   - The negative-cost and null-key checks are intentionally left **unchanged** — they remain unconditional failures, because they are *output-contract violations* (the function returned an invalid value) rather than *user-function exceptions*. This matches the sibling `groupedWeighted`/`groupedWeightedWithin` operators.
   - `decider` is a `lazy val` → zero overhead on the happy path.
   - Document supervision adherence on both operators in the Scala/Java DSL scaladoc and the operator reference pages.
   
   ### Result
   
   `groupedAdjacentBy` and `groupedAdjacentByWeighted` now adhere to the `SupervisionStrategy` attribute for both user functions.
   
   ### Tests
   
   - `sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowGroupedAdjacentByWeightedSpec"` — **14/14 passed**, including 8 new directional tests covering both functions: `costFn` throws (Stop, default Stop, Resume, Restart), key function throws (Stop, Resume, Restart, and Resume-on-last-element flushed at completion).
   - `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: notifications-unsubscribe@pekko.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org

Re: [PR] feat: add supervision strategy support for groupedAdjacentBy/Weighted [pekko]

["He-Pin (via GitHub)" <gi...@apache.org>]

He-Pin merged PR #3181:
URL: https://github.com/apache/pekko/pull/3181


-- 
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: notifications-unsubscribe@pekko.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org