feat: add supervision strategy support for groupedAdjacentBy/Weighted#3181
Open
He-Pin wants to merge 2 commits into
Open
feat: add supervision strategy support for groupedAdjacentBy/Weighted#3181He-Pin wants to merge 2 commits into
He-Pin wants to merge 2 commits into
Conversation
Motivation: The groupedAdjacentBy and groupedAdjacentByWeighted operators apply two user-provided functions, a key function and a cost function, but both were called without a try-catch. Any exception failed the stream unconditionally, ignoring the configured SupervisionStrategy. Modification: Wrap both the costFn and the key function calls in GroupedAdjacentByWeighted with a try-catch that consults the SupervisionStrategy decider via a shared superviseUserFn helper. Stop fails the stage, Resume skips the offending element while keeping the current group, and Restart drops the in-progress group. The negative-cost and null-key contract checks are left unchanged (they remain unconditional failures, since they are output-contract violations rather than user-function exceptions). The decider is a lazy val for zero overhead on the happy path. Update the Scala and Java DSL scaladoc for both operators 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 References: Refs apache#3110
Motivation: The weighted variant's documentation explicitly describes what Resume and Restart do, but the non-weighted groupedAdjacentBy only said "adheres to SupervisionStrategy" without the detail. Modification: Extend the supervision note in groupedAdjacentBy.md and the groupedAdjacentBy Scaladoc on FlowOps and all four Java DSL classes (Flow, Source, SubFlow, SubSource) to state that Resume skips the offending element while Restart drops the current in-progress group, matching the weighted variant. Result: Both groupedAdjacentBy operators document their supervision semantics with the same level of detail across all DSLs. Tests: - sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowGroupedAdjacentByWeightedSpec" (14/14 passed) References: Refs apache#3110
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Motivation
Per the stream error handling docs, operators that apply user-provided functions should consult the configured
SupervisionStrategy. ThegroupedAdjacentByandgroupedAdjacentByWeightedoperators apply two user functions — a key functionfand (for the weighted variant) acostFn— but both were called without a try-catch, so any exception failed the stream unconditionally, even underSupervision.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
costFnand the key function calls inGroupedAdjacentByWeightedwith atry/catch(NonFatal) that consults theSupervisionStrategydecider via a sharedsuperviseUserFnhelper:groupedWeighted/groupedWeightedWithinoperators.decideris alazy val→ zero overhead on the happy path.Result
groupedAdjacentByandgroupedAdjacentByWeightednow adhere to theSupervisionStrategyattribute 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:costFnthrows (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.