This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch feat/stream-expand-supervision in repository https://gitbox.apache.org/repos/asf/pekko.git
commit 35b7281cb64c0c22f15e41e9dbfee8fdf3232754 Author: 虎鸣 <[email protected]> AuthorDate: Sat Jun 27 03:27:23 2026 +0800 fix: align Expand decider and clarify iterator-failure supervision semantics Motivation: Review of the expand/extrapolate supervision support identified two consistency and documentation gaps. Modification: - Change `private lazy val decider` to `private def decider` in Expand to match the established pattern in Map, Filter, Collect and other fusing operators. - Collapse the Resume and Restart branches in `handleIteratorFailure` since a corrupt iterator must be discarded regardless of the supervision decision, and add a comment explaining why. - Clarify in expand.md that iterator evaluation failures necessarily discard extrapolation state under Resume, while expander-function failures preserve it. Result: Expand's supervision handling is consistent with neighbouring operators and its docs disambiguate the two failure sites. Tests: - sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowExpandSpec org.apache.pekko.stream.scaladsl.FlowExtrapolateSpec" (23/23 passed) References: Refs #3110 --- docs/src/main/paradox/stream/operators/Source-or-Flow/expand.md | 8 +++++--- .../src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala | 9 +++------ 2 files changed, 8 insertions(+), 9 deletions(-) diff --git a/docs/src/main/paradox/stream/operators/Source-or-Flow/expand.md b/docs/src/main/paradox/stream/operators/Source-or-Flow/expand.md index 3709833314..6dab793c98 100644 --- a/docs/src/main/paradox/stream/operators/Source-or-Flow/expand.md +++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/expand.md @@ -18,9 +18,11 @@ See @ref:[Understanding extrapolate and expand](../../stream-rate.md#understandi and examples. This operator adheres to the `ActorAttributes.SupervisionStrategy` attribute for exceptions thrown by the `expander` -function. On `Supervision.Stop` the stream fails; on `Supervision.Resume` the failed element is dropped and current -extrapolation state is kept when available; on `Supervision.Restart` the failed element is dropped and the current -extrapolation state is reset. +function or during iterator evaluation (`hasNext`/`next`). On `Supervision.Stop` the stream fails; on +`Supervision.Resume` the failed element is dropped and the current extrapolation state is kept when the failure +occurred in the `expander` function (a previously active iterator is retained), but is necessarily discarded when +the failure occurred during iterator evaluation; on `Supervision.Restart` the failed element is dropped and the +current extrapolation state is reset. ## Example diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala index 5529539847..6f55ee08be 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala @@ -1200,7 +1200,7 @@ private[stream] object Collect { override val shape = FlowShape(in, out) override def createLogic(attr: Attributes) = new GraphStageLogic(shape) with InHandler with OutHandler { - private lazy val decider = attr.mandatoryAttribute[SupervisionStrategy].decider + private def decider = attr.mandatoryAttribute[SupervisionStrategy].decider private var iterator: Iterator[Out] = Iterator.empty private var expanded = false private val contextPropagation = ContextPropagation() @@ -1276,11 +1276,8 @@ private[stream] object Collect { decider(ex) match { case Supervision.Stop => failStage(ex) - case Supervision.Resume => - restartState() - if (isClosed(in)) completeStage() - else if (!hasBeenPulled(in)) pull(in) - case Supervision.Restart => + case Supervision.Resume | Supervision.Restart => + // iterator is corrupt after any hasNext/next failure; must reset for both Resume and Restart restartState() if (isClosed(in)) completeStage() else if (!hasBeenPulled(in)) pull(in) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
