The GitHub Actions job "Pull Requests" on 
pekko.git/feat/stream-groupedweighted-supervision has succeeded.
Run started by GitHub user He-Pin (triggered by He-Pin).

Head commit for run:
953da3a9f58d58ad6d7156299fb5337cb064f4aa / 虎鸣 <[email protected]>
feat: add supervision strategy support for groupedWeighted costFn

Motivation:
The groupedWeighted operator's costFn is user-provided and may throw, but it
was called without a try-catch. Any exception failed the stream
unconditionally, ignoring the configured SupervisionStrategy. This was
inconsistent with operators such as map, filter and batch which already honor
supervision.

Modification:
Wrap costFn(elem) in GroupedWeighted.onPush() with a try-catch that consults
the SupervisionStrategy decider:
- Stop fails the stage (unchanged fail-fast behavior).
- Resume skips the offending element and keeps the current partial group.
- Restart drops the accumulated group and resets the weight counter.
The decider is a lazy val so there is zero overhead on the happy path. Also
add SourceLocation.forLambda(costFn) to initialAttributes for better failure
attribution, matching the sibling groupedWeightedWithin. Update the Scala and
Java DSL scaladoc and the operator reference page to document supervision
adherence.

Result:
groupedWeighted now adheres to the SupervisionStrategy attribute. Streams
configured with Resume or Restart survive costFn exceptions with well-defined
group semantics, consistent with the rest of the stream operators.

Tests:
- sbt "stream-tests/Test/testOnly 
org.apache.pekko.stream.scaladsl.FlowGroupedWeightedSpec" -- 14/14 passed

References:
Refs #3110

Report URL: https://github.com/apache/pekko/actions/runs/28147088807

With regards,
GitHub Actions via GitBox


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to