fix: add supervision strategy support for zipWith and zipWithN zipper functions#3186
Open
He-Pin wants to merge 2 commits into
Open
fix: add supervision strategy support for zipWith and zipWithN zipper functions#3186He-Pin wants to merge 2 commits into
He-Pin wants to merge 2 commits into
Conversation
… functions Motivation: zipWith and zipWithN accepted user zipper/combiner functions but did not consult SupervisionStrategy. Zipper exceptions failed the stream unconditionally, even when Resume/Restart was configured. Modification: Add zipper exception supervision handling in both implementations: - ZipWith boilerplate template (all arities) - Graph.ZipWithN Stop fails the stage; Resume/Restart drop the failing zipped element and continue by pulling new inputs. Pending bookkeeping is updated on resumed/ restarted failures so the stage can satisfy the current downstream demand without stalling. Update Scala/Java API docs and operator docs (`zipWith.md`, `zipWithN.md`) to document supervision behavior for combiner function failures. Add directional tests in: - FlowZipWithSpec - GraphZipWithNSpec covering explicit Stop, default Stop, Resume, and Restart. Result: zipWith/zipWithN now honor ActorAttributes.SupervisionStrategy for zipper failures with deterministic regression coverage for dropped-element paths. Tests: - sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowZipWithSpec org.apache.pekko.stream.scaladsl.GraphZipWithNSpec" -- 37/37 passed - sbt "stream/mimaReportBinaryIssues" -- clean - sbt "docs/paradox" -- passed References: Refs apache#3110
Motivation: The supervision support added in the previous commit causes ZipWith zipper exceptions to be handled by the decider instead of being logged as ERROR. GraphZipWithSpec still wrapped its sad-case test in EventFilter[ArithmeticException], which now times out waiting for an ERROR log message that is never emitted. Modification: Remove the EventFilter wrapping and unused import from GraphZipWithSpec's "work in the sad case" test, matching the same cleanup already applied to FlowZipWithSpec and GraphZipWithNSpec. Result: GraphZipWithSpec sad-case test passes without EventFilter timeout. Tests: - sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.GraphZipWithSpec" -- 15/15 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
Issue #3110 tracks stream operators that should honor
SupervisionStrategyfor user functions.zipWith/zipWithNinvoke user-provided zipper functions, but failures were not supervised: exceptions failed the stream unconditionally.Modification
ZipWithApply.scala.template(covers allZipWitharities)Graph.scala(ZipWithN)Stop→ fail stageResume/Restart→ drop failing zipped element and continuezipWith.mdzipWithN.mdFlowZipWithSpecGraphZipWithNSpeccovering explicit Stop, default Stop, Resume, Restart.
Result
zipWithandzipWithNnow honorActorAttributes.SupervisionStrategyfor zipper function failures, with regression coverage for dropped-element paths.which is now Apache licensed
Tests
sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowZipWithSpec org.apache.pekko.stream.scaladsl.GraphZipWithNSpec"(37/37 passed)sbt "stream/mimaReportBinaryIssues"(clean)sbt "docs/paradox"(passed)References
Refs #3110