feat: add supervision strategy support for doOnFirst#3179
Open
He-Pin wants to merge 2 commits into
Open
Conversation
Motivation: The doOnFirst operator runs a user-provided side-effect function on the first element, but it was called without a try-catch. Any exception failed the stream unconditionally, ignoring the configured SupervisionStrategy. Modification: Wrap f(elem) in DoOnFirst with a try-catch that consults the SupervisionStrategy decider. Stop fails the stage (unchanged), Resume drops the failing first element and switches to pass-through without retrying the function, and Restart re-arms first-element detection so the next element is treated as the first. The decider is a lazy val for zero overhead on the happy path. The first-element handler was extracted into a named val so it can be re-armed on Restart. Also fix a fully-qualified type reference in the createLogic signature to use the project's two-line import style, and update the Scala/Java DSL scaladoc and the operator reference page. Result: doOnFirst now adheres to the SupervisionStrategy attribute with clear, distinguishable Resume and Restart semantics. Tests: - sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowDoOnFirstSpec" -- 6/6 passed References: Refs apache#3110
Motivation: The existing restart test only covered the path where f throws for every element. The more common restart scenario is a transient failure on the first element followed by success, after which the operator must switch to pass-through. Modification: Add "restart and continue with pass-through once f succeeds on retry" which asserts that the second element is treated as the first (f runs again), and that once f succeeds the operator transitions to the pass-through handler so later elements do not re-invoke f. Result: The doOnFirst restart semantics are now covered for both the all-failing and the succeed-on-retry paths. Tests: - sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowDoOnFirstSpec" (7/7 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. ThedoOnFirstoperator runs a user side-effect functionfon the first element, but it was 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 per PR.
Modification
f(elem)inDoOnFirstwith atry/catch(NonFatal) that consults theSupervisionStrategydecider:fis not retried (the one-shot is consumed)fis retried on it)InHandlerwas extracted into a namedval firstHandlerso it can be re-armed on Restart. On success, the switch to pass-through now happens afterf(behaviorally identical on the happy path —pushdoes not synchronously re-enteronPush).decideris alazy val→ zero overhead on the happy path.createLogicsignature to use the project's two-line import style.Result
doOnFirstnow adheres to theSupervisionStrategyattribute with clear, distinguishable Resume (drop first, no retry) and Restart (next becomes first) semantics.Tests
sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowDoOnFirstSpec"— 6/6 passed, including 4 new directional tests:fthrow withSupervision.StopResumedrops the first element, does not retryf, passes the rest throughRestartre-arms sofis retried on each subsequent "first" elementsbt "stream/mimaReportBinaryIssues"— clean (binary compatible)References
Refs #3110
This is a clean-room implementation written directly for Apache Pekko.