This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 84aa2b20e4 fix: add supervision strategy support for expand and
extrapolate (#3185)
84aa2b20e4 is described below
commit 84aa2b20e481d5fec069b344750005732b6bec8e
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Mon Jun 29 05:36:48 2026 +0800
fix: add supervision strategy support for expand and extrapolate (#3185)
Motivation:
Issue #3110 tracks stream operators that should honor
ActorAttributes.SupervisionStrategy for user functions. The expand
operator evaluated user-provided expansion logic without supervision,
so exceptions failed the stream unconditionally. Because extrapolate is
implemented on top of Expand, the same gap applied there.
Modification:
Add supervision handling in Expand for exceptions from the expander
function and from iterator evaluation paths (hasNext/next). Stop fails
the stage, Resume drops the failed element and continues, and Restart
resets extrapolation state before continuing. A corrupt iterator (any
hasNext/next failure) is necessarily discarded under both Resume and
Restart since its state cannot be trusted.
Change the decider accessor to `private def decider` to match the
established pattern in Map, Filter, Collect, and other fusing operators.
Update Scala/Java API docs (Flow/Source/SubFlow/SubSource) and operator
docs (expand.md/extrapolate.md) to describe supervision behavior,
including iterator evaluation failures. Fix a long-standing Scaladoc
typo ("faster than the upstream" → "faster than the downstream") in the
expand documentation, and clarify in extrapolate's docs that Resume and
Restart behave identically because the original element is always
emitted before the extrapolator is invoked.
Add directional tests in FlowExpandSpec (15 tests, +10) and
FlowExtrapolateSpec (11 tests, +4) covering Stop/default-Stop/Resume/
Restart for both expander-function failures and iterator-evaluation
failures, including regression tests for iterator failures that occur
during upstream completion (onUpstreamFinish).
Result:
expand and extrapolate now honor ActorAttributes.SupervisionStrategy for
user-function and iterator-evaluation failures, with deterministic
regression coverage for all supervision branches.
Tests:
- sbt "stream-tests/Test/testOnly
org.apache.pekko.stream.scaladsl.FlowExpandSpec
org.apache.pekko.stream.scaladsl.FlowExtrapolateSpec" -- 26/26 passed
- sbt "stream/mimaReportBinaryIssues" -- clean (unchanged from prior
commits)
- sbt "docs/paradox" -- passed (unchanged from prior commits)
References:
Refs #3110
---
.../stream/operators/Source-or-Flow/expand.md | 8 +-
.../stream/operators/Source-or-Flow/extrapolate.md | 6 +
.../pekko/stream/scaladsl/FlowExpandSpec.scala | 197 +++++++++++++++++++++
.../stream/scaladsl/FlowExtrapolateSpec.scala | 108 +++++++++++
.../org/apache/pekko/stream/impl/fusing/Ops.scala | 118 ++++++++----
.../org/apache/pekko/stream/javadsl/Flow.scala | 37 +++-
.../org/apache/pekko/stream/javadsl/Source.scala | 37 +++-
.../org/apache/pekko/stream/javadsl/SubFlow.scala | 37 +++-
.../apache/pekko/stream/javadsl/SubSource.scala | 37 +++-
.../org/apache/pekko/stream/scaladsl/Flow.scala | 27 ++-
10 files changed, 531 insertions(+), 81 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 d1ea059226..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
@@ -17,6 +17,13 @@ element, allowing for it to be rewritten and/or filtered.
See @ref:[Understanding extrapolate and
expand](../../stream-rate.md#understanding-extrapolate-and-expand) for more
information
and examples.
+This operator adheres to the `ActorAttributes.SupervisionStrategy` attribute
for exceptions thrown by the `expander`
+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
Imagine a streaming client decoding a video. It is possible the network
bandwidth is a bit
@@ -43,4 +50,3 @@ Java
**completes** when upstream completes
@@@
-
diff --git
a/docs/src/main/paradox/stream/operators/Source-or-Flow/extrapolate.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/extrapolate.md
index 97ca17edab..6ebf6fb508 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/extrapolate.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/extrapolate.md
@@ -22,6 +22,12 @@ Includes an optional `initial` argument to prevent blocking
the entire stream wh
See @ref:[Understanding extrapolate and
expand](../../stream-rate.md#understanding-extrapolate-and-expand) for more
information
and examples.
+This operator adheres to the `ActorAttributes.SupervisionStrategy` attribute
for exceptions thrown by the `extrapolator`
+function or during iterator evaluation (`hasNext`/`next`). On
`Supervision.Stop` the stream fails; on
+`Supervision.Resume` the failed element is dropped and any previously active
extrapolation is retained; on
+`Supervision.Restart` the failed element is dropped and the current
extrapolation state is reset. For
+iterator-evaluation failures, `Resume` and `Restart` both discard the corrupt
iterator because it cannot be reused.
+
## Example
Imagine a videoconference client decoding a video feed from a colleague
working remotely. It is possible
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowExpandSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowExpandSpec.scala
index 56e8accff8..3b958c538c 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowExpandSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowExpandSpec.scala
@@ -20,6 +20,7 @@ import scala.concurrent.duration._
import org.apache.pekko
import pekko.stream.ActorAttributes
+import pekko.stream.Supervision
import pekko.stream.testkit._
import pekko.stream.testkit.scaladsl.TestSink
import pekko.stream.testkit.scaladsl.TestSource
@@ -150,6 +151,202 @@ class FlowExpandSpec extends StreamSpec("""
source.sendNext(2).sendComplete()
sink.expectNext(2 -> 0).expectComplete()
}
+
+ "fail stream when expander throws and supervision is Stop" in {
+ val ex = new RuntimeException("boom")
+ val result = Source(1 to 5)
+ .expand(i => if (i == 3) throw ex else Iterator.single(i))
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.stoppingDecider))
+ .runWith(Sink.ignore)
+ result.failed.futureValue shouldBe ex
+ }
+
+ "fail stream when expander throws and supervision defaults to Stop" in {
+ val ex = new RuntimeException("boom")
+ val result = Source(1 to 5)
+ .expand(i => if (i == 3) throw ex else Iterator.single(i))
+ .runWith(Sink.ignore)
+ result.failed.futureValue shouldBe ex
+ }
+
+ "resume and keep current extrapolation when expander throws" in {
+ val ex = new RuntimeException("boom")
+ val publisher = TestPublisher.probe[Int]()
+ val subscriber = TestSubscriber.probe[Int]()
+
+ Source
+ .fromPublisher(publisher)
+ .expand(i => if (i == 1) Iterator(1, 10, 11) else if (i == 2) throw ex
else Iterator.single(i))
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
+ .to(Sink.fromSubscriber(subscriber))
+ .run()
+
+ publisher.sendNext(1)
+ subscriber.requestNext(1)
+ publisher.sendNext(2) // throws in expander, element 2 is dropped under
Resume
+ subscriber.requestNext(10) // continue from current extrapolation
+ subscriber.cancel()
+ }
+
+ "complete immediately on upstream finish if expanded is true" in {
+ val subscriber = TestSubscriber.probe[Int]()
+
+ Source.single(1)
+ .expand(_ => Iterator(1, 2, 3))
+ .to(Sink.fromSubscriber(subscriber))
+ .run()
+
+ subscriber.requestNext(1) // expanded becomes true
+ subscriber.expectComplete() // must complete immediately when expanded
is true
+ }
+
+ "restart and reset current extrapolation when expander throws" in {
+ val ex = new RuntimeException("boom")
+ val publisher = TestPublisher.probe[Int]()
+ val subscriber = TestSubscriber.probe[Int]()
+
+ Source
+ .fromPublisher(publisher)
+ .expand(i => if (i == 1) Iterator(1, 10, 11) else if (i == 2) throw ex
else Iterator.single(i))
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.restartingDecider))
+ .to(Sink.fromSubscriber(subscriber))
+ .run()
+
+ publisher.sendNext(1)
+ subscriber.requestNext(1)
+ publisher.sendNext(2) // throws in expander, restart resets iterator
state
+ subscriber.request(1)
+ subscriber.expectNoMessage(300.millis)
+ publisher.sendNext(3)
+ subscriber.requestNext(3)
+ subscriber.cancel()
+ }
+
+ "resume when iterator produced by expander throws during iteration" in {
+ val ex = new RuntimeException("boom")
+ val result = Source(1 to 4)
+ .expand(i =>
+ if (i == 3)
+ new Iterator[Int] {
+ override def hasNext: Boolean = true
+ override def next(): Int = throw ex
+ }
+ else Iterator.single(i))
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
+ .runWith(Sink.seq)
+
+ Await.result(result, 3.seconds) shouldBe Seq(1, 2, 4)
+ }
+
+ "restart when iterator produced by expander throws during iteration" in {
+ val ex = new RuntimeException("boom")
+ val result = Source(1 to 4)
+ .expand(i =>
+ if (i == 3)
+ new Iterator[Int] {
+ override def hasNext: Boolean = true
+ override def next(): Int = throw ex
+ }
+ else Iterator.single(i))
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.restartingDecider))
+ .runWith(Sink.seq)
+
+ Await.result(result, 3.seconds) shouldBe Seq(1, 2, 4)
+ }
+
+ "fail stream when iterator throws during upstream completion" in {
+ val ex = new RuntimeException("boom")
+ val publisher = TestPublisher.probe[Int]()
+ val subscriber = TestSubscriber.probe[Int]()
+
+ Source
+ .fromPublisher(publisher)
+ .expand(_ =>
+ new Iterator[Int] {
+ private var calls = 0
+ override def hasNext: Boolean = {
+ calls += 1
+ if (calls >= 2) throw ex else true
+ }
+ override def next(): Int = 1
+ })
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.stoppingDecider))
+ .to(Sink.fromSubscriber(subscriber))
+ .run()
+
+ subscriber.request(1) // ensure isAvailable(out) during onPush so
hasNext #1 fires there
+ publisher.sendNext(1) // onPush: hasNext #1 true, pull
+ subscriber.expectNext(1)
+ publisher.sendComplete() // onUpstreamFinish: hasNext #2 throws ->
failStage
+ subscriber.expectError(ex)
+ }
+
+ "resume and complete when iterator throws during upstream completion with
no pending pull" in {
+ val ex = new RuntimeException("boom")
+ val publisher = TestPublisher.probe[Int]()
+ val subscriber = TestSubscriber.probe[Int]()
+
+ Source
+ .fromPublisher(publisher)
+ .expand(_ =>
+ new Iterator[Int] {
+ private var calls = 0
+ override def hasNext: Boolean = {
+ calls += 1
+ if (calls >= 2) throw ex else true
+ }
+ override def next(): Int = 1
+ })
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
+ .to(Sink.fromSubscriber(subscriber))
+ .run()
+
+ subscriber.request(1) // ensure isAvailable(out) during onPush so
hasNext #1 fires there
+ publisher.sendNext(1) // onPush: hasNext #1 true, pull
+ subscriber.expectNext(1)
+ publisher.sendComplete() // onUpstreamFinish: hasNext #2 throws ->
resume -> completeStage
+ subscriber.expectComplete()
+ }
+
+ "resume when iterator throws during onPull with expanded = true" in {
+ val ex = new RuntimeException("boom")
+ val (publisher, subscriber) = TestSource[Int]()
+ .expand(i =>
+ if (i == 2)
+ new Iterator[Int] {
+ private var calls = 0
+ override def hasNext: Boolean = {
+ calls += 1
+ // 1st hasNext (in onPush) succeeds -> element 2 pushed,
expanded=true
+ // 2nd hasNext (in onPull during extrapolation) throws
+ if (calls >= 2) throw ex else true
+ }
+ override def next(): Int = 2
+ }
+ else Iterator.single(i))
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
+ .toMat(TestSink[Int]())(Keep.both)
+ .run()
+
+ subscriber.request(1)
+ publisher.sendNext(1)
+ subscriber.expectNext(1)
+
+ // hasNext #1 succeeds in onPush, element 2 emitted, expanded=true
+ subscriber.request(1)
+ publisher.sendNext(2)
+ subscriber.expectNext(2)
+
+ // hasNext #2 throws in onPull; Resume resets state, pulls next
+ subscriber.request(1)
+
+ // Element 3 flows through normally
+ publisher.sendNext(3)
+ subscriber.expectNext(3)
+
+ publisher.sendComplete()
+ subscriber.expectComplete()
+ }
}
}
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowExtrapolateSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowExtrapolateSpec.scala
index d3dbc7eebf..e288a27185 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowExtrapolateSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowExtrapolateSpec.scala
@@ -19,6 +19,8 @@ import scala.concurrent.Await
import scala.concurrent.duration._
import org.apache.pekko
+import pekko.stream.ActorAttributes
+import pekko.stream.Supervision
import pekko.stream.testkit._
import pekko.stream.testkit.scaladsl.TestSink
import pekko.stream.testkit.scaladsl.TestSource
@@ -170,6 +172,112 @@ class FlowExtrapolateSpec extends StreamSpec("""
source.sendNext(2).sendComplete()
sink.expectNext(2 -> 0).expectComplete()
}
+
+ "stop when extrapolator throws during iterator evaluation" in {
+ val ex = new RuntimeException("boom")
+ val publisher = TestPublisher.probe[Int]()
+ val subscriber = TestSubscriber.probe[Int]()
+
+ Source
+ .fromPublisher(publisher)
+ .extrapolate(i =>
+ if (i == 3)
+ new Iterator[Int] {
+ override def hasNext: Boolean = throw ex
+ override def next(): Int = throw ex
+ }
+ else Iterator.empty)
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.stoppingDecider))
+ .to(Sink.fromSubscriber(subscriber))
+ .run()
+
+ publisher.sendNext(3)
+ subscriber.requestNext(3)
+ subscriber.request(1)
+ subscriber.expectError(ex)
+ }
+
+ "resume when extrapolator throws during iterator evaluation" in {
+ val ex = new RuntimeException("boom")
+ val publisher = TestPublisher.probe[Int]()
+ val subscriber = TestSubscriber.probe[Int]()
+
+ Source
+ .fromPublisher(publisher)
+ .extrapolate(i =>
+ if (i == 3)
+ new Iterator[Int] {
+ override def hasNext: Boolean = throw ex
+ override def next(): Int = throw ex
+ }
+ else Iterator.empty)
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
+ .to(Sink.fromSubscriber(subscriber))
+ .run()
+
+ publisher.sendNext(3)
+ subscriber.requestNext(3)
+ subscriber.request(1)
+ subscriber.expectNoMessage(300.millis)
+ publisher.sendNext(4)
+ subscriber.requestNext(4)
+ subscriber.cancel()
+ }
+
+ "restart when extrapolator throws during iterator evaluation" in {
+ val ex = new RuntimeException("boom")
+ val publisher = TestPublisher.probe[Int]()
+ val subscriber = TestSubscriber.probe[Int]()
+
+ Source
+ .fromPublisher(publisher)
+ .extrapolate(i =>
+ if (i == 3)
+ new Iterator[Int] {
+ override def hasNext: Boolean = throw ex
+ override def next(): Int = throw ex
+ }
+ else Iterator.empty)
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.restartingDecider))
+ .to(Sink.fromSubscriber(subscriber))
+ .run()
+
+ publisher.sendNext(3)
+ subscriber.requestNext(3)
+ subscriber.request(1)
+ subscriber.expectNoMessage(300.millis)
+ publisher.sendNext(4)
+ subscriber.requestNext(4)
+ subscriber.cancel()
+ }
+
+ "fail stream when extrapolator iterator throws during upstream completion"
in {
+ val ex = new RuntimeException("boom")
+ val publisher = TestPublisher.probe[Int]()
+ val subscriber = TestSubscriber.probe[Int]()
+
+ Source
+ .fromPublisher(publisher)
+ .extrapolate(_ =>
+ new Iterator[Int] {
+ private var calls = 0
+ override def hasNext: Boolean = {
+ calls += 1
+ if (calls >= 2) throw ex else true
+ }
+ override def next(): Int = 99
+ })
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.stoppingDecider))
+ .to(Sink.fromSubscriber(subscriber))
+ .run()
+
+ subscriber.request(2) // demand for original element + one extrapolated
+ publisher.sendNext(1)
+ subscriber.expectNext(1) // original element (Iterator.single(1) ++
extrapolator(1))
+ subscriber.expectNext(99) // first extrapolated element -> hasNext #1
true, next returns 99
+ publisher.sendComplete() // onUpstreamFinish -> hasNext #2 throws ->
failStage
+ subscriber.expectError(ex)
+ }
}
}
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 fc9f3e1dda..4c0bf7bb93 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
@@ -1226,50 +1226,94 @@ private[stream] object Collect {
override val shape = FlowShape(in, out)
- override def createLogic(attr: Attributes) = new GraphStageLogic(shape) with
InHandler with OutHandler {
- private var iterator: Iterator[Out] = Iterator.empty
- private var expanded = false
- private val contextPropagation = ContextPropagation()
+ override def createLogic(inheritedAttributes: Attributes) =
+ new GraphStageLogic(shape) with InHandler with OutHandler {
+ private lazy val decider =
inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider
+ private var iterator: Iterator[Out] = Iterator.empty
+ private var expanded = false
+ private val contextPropagation = ContextPropagation()
- override def preStart(): Unit = pull(in)
+ override def preStart(): Unit = pull(in)
- def onPush(): Unit = {
- iterator = extrapolate(grab(in))
- if (iterator.hasNext) {
- contextPropagation.suspendContext()
- if (isAvailable(out)) {
- expanded = true
- pull(in)
- push(out, iterator.next())
- } else expanded = false
- } else pull(in)
- }
+ def onPush(): Unit = {
+ val elem = grab(in)
+ try iterator = extrapolate(elem)
+ catch {
+ case NonFatal(ex) =>
+ decider(ex) match {
+ case Supervision.Stop =>
+ failStage(ex)
+ case Supervision.Resume =>
+ if (!isClosed(in) && !hasBeenPulled(in)) pull(in)
+ case Supervision.Restart =>
+ restartState()
+ if (!isClosed(in) && !hasBeenPulled(in)) pull(in)
+ }
+ return
+ }
+ try {
+ if (iterator.hasNext) {
+ contextPropagation.suspendContext()
+ if (isAvailable(out)) {
+ expanded = true
+ pull(in)
+ push(out, iterator.next())
+ } else expanded = false
+ } else pull(in)
+ } catch {
+ case NonFatal(ex) => handleIteratorFailure(ex)
+ }
+ }
- override def onUpstreamFinish(): Unit = {
- if (iterator.hasNext && !expanded) () // need to wait
- else completeStage()
- }
+ override def onUpstreamFinish(): Unit = {
+ try {
+ if (iterator.hasNext && !expanded) () // need to wait
+ else completeStage()
+ } catch {
+ case NonFatal(ex) => handleIteratorFailure(ex)
+ }
+ }
- def onPull(): Unit = {
- if (iterator.hasNext) {
- contextPropagation.resumeContext()
- if (!expanded) {
- expanded = true
- if (isClosed(in)) {
- push(out, iterator.next())
- completeStage()
- } else {
- // expand needs to pull first to be “fair” when upstream is not
actually slow
- pull(in)
- push(out, iterator.next())
+ def onPull(): Unit = {
+ try {
+ if (iterator.hasNext) {
+ contextPropagation.resumeContext()
+ if (!expanded) {
+ expanded = true
+ if (isClosed(in)) {
+ push(out, iterator.next())
+ completeStage()
+ } else {
+ // expand needs to pull first to be “fair” when upstream is
not actually slow
+ pull(in)
+ push(out, iterator.next())
+ }
+ } else push(out, iterator.next())
}
- } else push(out, iterator.next())
+ } catch {
+ case NonFatal(ex) => handleIteratorFailure(ex)
+ }
}
- }
- setHandler(in, this)
- setHandler(out, this)
- }
+ private def restartState(): Unit = {
+ iterator = Iterator.empty
+ expanded = false
+ }
+
+ private def handleIteratorFailure(ex: Throwable): Unit =
+ decider(ex) match {
+ case Supervision.Stop =>
+ failStage(ex)
+ 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)
+ }
+
+ setHandler(in, this)
+ setHandler(out, this)
+ }
}
/**
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
index 44e1f4c17f..50c8e2a0e2 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
@@ -2587,12 +2587,19 @@ final class Flow[In, Out, Mat](delegate:
scaladsl.Flow[In, Out, Mat]) extends Gr
* element until new element comes from the upstream. For example an expand
step might repeat the last element for
* the subscriber until it receives an update from upstream.
*
- * This element will never "drop" upstream elements as all elements go
through at least one extrapolation step.
- * This means that if the upstream is actually faster than the upstream it
will be backpressured by the downstream
- * subscriber.
+ * Under normal operation all upstream elements go through at least one
extrapolation step.
+ * If supervision drops a failed element, that element is not emitted.
+ * If the upstream is actually faster than the downstream it will be
backpressured by the downstream subscriber.
*
- * Expand does not support [[pekko.stream.Supervision#restart]] and
[[pekko.stream.Supervision#resume]].
- * Exceptions from the `expander` function will complete the stream with
failure.
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the `expander` function or the iterator it produces throws during
evaluation (`hasNext`/`next`) and the
+ * supervision decision is [[pekko.stream.Supervision#stop]] the stream
fails. If the supervision decision is
+ * [[pekko.stream.Supervision#resume]] the failed element is dropped and the
stream continues, keeping current
+ * extrapolation state when available. If the supervision decision is
[[pekko.stream.Supervision#restart]] the
+ * failed element is dropped and the current extrapolation state is reset.
Note that iterator-evaluation failures
+ * necessarily discard the current iterator under both `resume` and
`restart`, because a corrupt iterator cannot
+ * be reused.
*
* See also [[#extrapolate]] for a version that always preserves the
original element and allows for an initial "startup" element.
*
@@ -2617,8 +2624,14 @@ final class Flow[In, Out, Mat](delegate:
scaladsl.Flow[In, Out, Mat]) extends Gr
* This is achieved by introducing "extrapolated" elements - based on those
from upstream - whenever downstream
* signals demand.
*
- * Extrapolate does not support [[pekko.stream.Supervision#restart]] and
[[pekko.stream.Supervision#resume]].
- * Exceptions from the `extrapolate` function will complete the stream with
failure.
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the `extrapolator` function or the iterator it produces throws during
evaluation (`hasNext`/`next`) and the
+ * supervision decision is [[pekko.stream.Supervision#stop]] the stream
fails. If the supervision decision is
+ * [[pekko.stream.Supervision#resume]] the failed element is dropped and any
previously active extrapolation is
+ * retained. If the supervision decision is
[[pekko.stream.Supervision#restart]] the failed element is dropped and
+ * the current extrapolation state is reset. For iterator-evaluation
failures, `resume` and `restart` both discard
+ * the corrupt iterator because it cannot be reused.
*
* See also [[#expand]] for a version that can overwrite the original
element.
*
@@ -2645,8 +2658,14 @@ final class Flow[In, Out, Mat](delegate:
scaladsl.Flow[In, Out, Mat]) extends Gr
* This is achieved by introducing "extrapolated" elements - based on those
from upstream - whenever downstream
* signals demand.
*
- * Extrapolate does not support [[pekko.stream.Supervision#restart]] and
[[pekko.stream.Supervision#resume]].
- * Exceptions from the `extrapolate` function will complete the stream with
failure.
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the `extrapolator` function or the iterator it produces throws during
evaluation (`hasNext`/`next`) and the
+ * supervision decision is [[pekko.stream.Supervision#stop]] the stream
fails. If the supervision decision is
+ * [[pekko.stream.Supervision#resume]] the failed element is dropped and any
previously active extrapolation is
+ * retained. If the supervision decision is
[[pekko.stream.Supervision#restart]] the failed element is dropped and
+ * the current extrapolation state is reset. For iterator-evaluation
failures, `resume` and `restart` both discard
+ * the corrupt iterator because it cannot be reused.
*
* See also [[#expand]] for a version that can overwrite the original
element.
*
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
index 2a8bce50f7..4c8923a6fe 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
@@ -4002,12 +4002,19 @@ final class Source[Out, Mat](delegate:
scaladsl.Source[Out, Mat]) extends Graph[
* element until new element comes from the upstream. For example an expand
step might repeat the last element for
* the subscriber until it receives an update from upstream.
*
- * This element will never "drop" upstream elements as all elements go
through at least one extrapolation step.
- * This means that if the upstream is actually faster than the upstream it
will be backpressured by the downstream
- * subscriber.
+ * Under normal operation all upstream elements go through at least one
extrapolation step.
+ * If supervision drops a failed element, that element is not emitted.
+ * If the upstream is actually faster than the downstream it will be
backpressured by the downstream subscriber.
*
- * Expand does not support [[pekko.stream.Supervision#restart]] and
[[pekko.stream.Supervision#resume]].
- * Exceptions from the `expander` function will complete the stream with
failure.
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the `expander` function or the iterator it produces throws during
evaluation (`hasNext`/`next`) and the
+ * supervision decision is [[pekko.stream.Supervision#stop]] the stream
fails. If the supervision decision is
+ * [[pekko.stream.Supervision#resume]] the failed element is dropped and the
stream continues, keeping current
+ * extrapolation state when available. If the supervision decision is
[[pekko.stream.Supervision#restart]] the
+ * failed element is dropped and the current extrapolation state is reset.
Note that iterator-evaluation failures
+ * necessarily discard the current iterator under both `resume` and
`restart`, because a corrupt iterator cannot
+ * be reused.
*
* See also [[#extrapolate]] for a version that always preserves the
original element and allows for an initial "startup" element.
*
@@ -4032,8 +4039,14 @@ final class Source[Out, Mat](delegate:
scaladsl.Source[Out, Mat]) extends Graph[
* This is achieved by introducing "extrapolated" elements - based on those
from upstream - whenever downstream
* signals demand.
*
- * Extrapolate does not support [[pekko.stream.Supervision#restart]] and
[[pekko.stream.Supervision#resume]].
- * Exceptions from the `extrapolate` function will complete the stream with
failure.
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the `extrapolator` function or the iterator it produces throws during
evaluation (`hasNext`/`next`) and the
+ * supervision decision is [[pekko.stream.Supervision#stop]] the stream
fails. If the supervision decision is
+ * [[pekko.stream.Supervision#resume]] the failed element is dropped and any
previously active extrapolation is
+ * retained. If the supervision decision is
[[pekko.stream.Supervision#restart]] the failed element is dropped and
+ * the current extrapolation state is reset. For iterator-evaluation
failures, `resume` and `restart` both discard
+ * the corrupt iterator because it cannot be reused.
*
* See also [[#expand]] for a version that can overwrite the original
element.
*
@@ -4060,8 +4073,14 @@ final class Source[Out, Mat](delegate:
scaladsl.Source[Out, Mat]) extends Graph[
* This is achieved by introducing "extrapolated" elements - based on those
from upstream - whenever downstream
* signals demand.
*
- * Extrapolate does not support [[pekko.stream.Supervision#restart]] and
[[pekko.stream.Supervision#resume]].
- * Exceptions from the `extrapolate` function will complete the stream with
failure.
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the `extrapolator` function or the iterator it produces throws during
evaluation (`hasNext`/`next`) and the
+ * supervision decision is [[pekko.stream.Supervision#stop]] the stream
fails. If the supervision decision is
+ * [[pekko.stream.Supervision#resume]] the failed element is dropped and any
previously active extrapolation is
+ * retained. If the supervision decision is
[[pekko.stream.Supervision#restart]] the failed element is dropped and
+ * the current extrapolation state is reset. For iterator-evaluation
failures, `resume` and `restart` both discard
+ * the corrupt iterator because it cannot be reused.
*
* See also [[#expand]] for a version that can overwrite the original
element.
*
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
index c5c3a2f7ca..189ad343a7 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
@@ -1912,12 +1912,19 @@ final class SubFlow[In, Out, Mat](
* element until new element comes from the upstream. For example an expand
step might repeat the last element for
* the subscriber until it receives an update from upstream.
*
- * This element will never "drop" upstream elements as all elements go
through at least one extrapolation step.
- * This means that if the upstream is actually faster than the upstream it
will be backpressured by the downstream
- * subscriber.
+ * Under normal operation all upstream elements go through at least one
extrapolation step.
+ * If supervision drops a failed element, that element is not emitted.
+ * If the upstream is actually faster than the downstream it will be
backpressured by the downstream subscriber.
*
- * Expand does not support [[pekko.stream.Supervision#restart]] and
[[pekko.stream.Supervision#resume]].
- * Exceptions from the `expander` function will complete the stream with
failure.
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the `expander` function or the iterator it produces throws during
evaluation (`hasNext`/`next`) and the
+ * supervision decision is [[pekko.stream.Supervision#stop]] the stream
fails. If the supervision decision is
+ * [[pekko.stream.Supervision#resume]] the failed element is dropped and the
stream continues, keeping current
+ * extrapolation state when available. If the supervision decision is
[[pekko.stream.Supervision#restart]] the
+ * failed element is dropped and the current extrapolation state is reset.
Note that iterator-evaluation failures
+ * necessarily discard the current iterator under both `resume` and
`restart`, because a corrupt iterator cannot
+ * be reused.
*
* See also [[#extrapolate]] for a version that always preserves the
original element and allows for an initial "startup" element.
*
@@ -1943,8 +1950,14 @@ final class SubFlow[In, Out, Mat](
* This is achieved by introducing "extrapolated" elements - based on those
from upstream - whenever downstream
* signals demand.
*
- * Extrapolate does not support [[pekko.stream.Supervision#restart]] and
[[pekko.stream.Supervision#resume]].
- * Exceptions from the `extrapolate` function will complete the stream with
failure.
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the `extrapolator` function or the iterator it produces throws during
evaluation (`hasNext`/`next`) and the
+ * supervision decision is [[pekko.stream.Supervision#stop]] the stream
fails. If the supervision decision is
+ * [[pekko.stream.Supervision#resume]] the failed element is dropped and any
previously active extrapolation is
+ * retained. If the supervision decision is
[[pekko.stream.Supervision#restart]] the failed element is dropped and
+ * the current extrapolation state is reset. For iterator-evaluation
failures, `resume` and `restart` both discard
+ * the corrupt iterator because it cannot be reused.
*
* See also [[#expand]] for a version that can overwrite the original
element.
*
@@ -1971,8 +1984,14 @@ final class SubFlow[In, Out, Mat](
* This is achieved by introducing "extrapolated" elements - based on those
from upstream - whenever downstream
* signals demand.
*
- * Extrapolate does not support [[pekko.stream.Supervision#restart]] and
[[pekko.stream.Supervision#resume]].
- * Exceptions from the `extrapolate` function will complete the stream with
failure.
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the `extrapolator` function or the iterator it produces throws during
evaluation (`hasNext`/`next`) and the
+ * supervision decision is [[pekko.stream.Supervision#stop]] the stream
fails. If the supervision decision is
+ * [[pekko.stream.Supervision#resume]] the failed element is dropped and any
previously active extrapolation is
+ * retained. If the supervision decision is
[[pekko.stream.Supervision#restart]] the failed element is dropped and
+ * the current extrapolation state is reset. For iterator-evaluation
failures, `resume` and `restart` both discard
+ * the corrupt iterator because it cannot be reused.
*
* See also [[#expand]] for a version that can overwrite the original
element.
*
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
index 86ad0103a2..ba572b0a86 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
@@ -1880,12 +1880,19 @@ final class SubSource[Out, Mat](
* element until new element comes from the upstream. For example an expand
step might repeat the last element for
* the subscriber until it receives an update from upstream.
*
- * This element will never "drop" upstream elements as all elements go
through at least one extrapolation step.
- * This means that if the upstream is actually faster than the upstream it
will be backpressured by the downstream
- * subscriber.
+ * Under normal operation all upstream elements go through at least one
extrapolation step.
+ * If supervision drops a failed element, that element is not emitted.
+ * If the upstream is actually faster than the downstream it will be
backpressured by the downstream subscriber.
*
- * Expand does not support [[pekko.stream.Supervision#restart]] and
[[pekko.stream.Supervision#resume]].
- * Exceptions from the `expander` function will complete the stream with
failure.
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the `expander` function or the iterator it produces throws during
evaluation (`hasNext`/`next`) and the
+ * supervision decision is [[pekko.stream.Supervision#stop]] the stream
fails. If the supervision decision is
+ * [[pekko.stream.Supervision#resume]] the failed element is dropped and the
stream continues, keeping current
+ * extrapolation state when available. If the supervision decision is
[[pekko.stream.Supervision#restart]] the
+ * failed element is dropped and the current extrapolation state is reset.
Note that iterator-evaluation failures
+ * necessarily discard the current iterator under both `resume` and
`restart`, because a corrupt iterator cannot
+ * be reused.
*
* See also [[#extrapolate]] for a version that always preserves the
original element and allows for an initial "startup" element.
*
@@ -1910,8 +1917,14 @@ final class SubSource[Out, Mat](
* This is achieved by introducing "extrapolated" elements - based on those
from upstream - whenever downstream
* signals demand.
*
- * Extrapolate does not support [[pekko.stream.Supervision#restart]] and
[[pekko.stream.Supervision#resume]].
- * Exceptions from the `extrapolate` function will complete the stream with
failure.
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the `extrapolator` function or the iterator it produces throws during
evaluation (`hasNext`/`next`) and the
+ * supervision decision is [[pekko.stream.Supervision#stop]] the stream
fails. If the supervision decision is
+ * [[pekko.stream.Supervision#resume]] the failed element is dropped and any
previously active extrapolation is
+ * retained. If the supervision decision is
[[pekko.stream.Supervision#restart]] the failed element is dropped and
+ * the current extrapolation state is reset. For iterator-evaluation
failures, `resume` and `restart` both discard
+ * the corrupt iterator because it cannot be reused.
*
* See also [[#expand]] for a version that can overwrite the original
element.
*
@@ -1938,8 +1951,14 @@ final class SubSource[Out, Mat](
* This is achieved by introducing "extrapolated" elements - based on those
from upstream - whenever downstream
* signals demand.
*
- * Extrapolate does not support [[pekko.stream.Supervision#restart]] and
[[pekko.stream.Supervision#resume]].
- * Exceptions from the `extrapolate` function will complete the stream with
failure.
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the `extrapolator` function or the iterator it produces throws during
evaluation (`hasNext`/`next`) and the
+ * supervision decision is [[pekko.stream.Supervision#stop]] the stream
fails. If the supervision decision is
+ * [[pekko.stream.Supervision#resume]] the failed element is dropped and any
previously active extrapolation is
+ * retained. If the supervision decision is
[[pekko.stream.Supervision#restart]] the failed element is dropped and
+ * the current extrapolation state is reset. For iterator-evaluation
failures, `resume` and `restart` both discard
+ * the corrupt iterator because it cannot be reused.
*
* See also [[#expand]] for a version that can overwrite the original
element.
*
diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
index 13e3338965..322108cd63 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
@@ -2522,12 +2522,19 @@ trait FlowOps[+Out, +Mat] {
* element until new element comes from the upstream. For example an expand
step might repeat the last element for
* the subscriber until it receives an update from upstream.
*
- * This element will never "drop" upstream elements as all elements go
through at least one extrapolation step.
- * This means that if the upstream is actually faster than the upstream it
will be backpressured by the downstream
- * subscriber.
+ * Under normal operation all upstream elements go through at least one
extrapolation step.
+ * If supervision drops a failed element, that element is not emitted.
+ * If the upstream is actually faster than the downstream it will be
backpressured by the downstream subscriber.
*
- * Expand does not support [[pekko.stream.Supervision.Restart]] and
[[pekko.stream.Supervision.Resume]].
- * Exceptions from the `seed` function will complete the stream with failure.
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the `expander` function or the iterator it produces throws during
evaluation (`hasNext`/`next`) and the
+ * supervision decision is [[pekko.stream.Supervision.Stop]] the stream
fails. If the supervision decision is
+ * [[pekko.stream.Supervision.Resume]] the failed element is dropped and the
stream continues, keeping current
+ * extrapolation state when available. If the supervision decision is
[[pekko.stream.Supervision.Restart]] the
+ * failed element is dropped and the current extrapolation state is reset.
Note that iterator-evaluation failures
+ * necessarily discard the current iterator under both `Resume` and
`Restart`, because a corrupt iterator cannot
+ * be reused.
*
* '''Emits when''' downstream stops backpressuring
*
@@ -2550,8 +2557,14 @@ trait FlowOps[+Out, +Mat] {
* This is achieved by introducing "extrapolated" elements - based on those
from upstream - whenever downstream
* signals demand.
*
- * Extrapolate does not support [[pekko.stream.Supervision.Restart]] and
[[pekko.stream.Supervision.Resume]].
- * Exceptions from the `extrapolate` function will complete the stream with
failure.
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the `extrapolator` function or the iterator it produces throws during
evaluation (`hasNext`/`next`) and the
+ * supervision decision is [[pekko.stream.Supervision.Stop]] the stream
fails. If the supervision decision is
+ * [[pekko.stream.Supervision.Resume]] the failed element is dropped and any
previously active extrapolation is
+ * retained. If the supervision decision is
[[pekko.stream.Supervision.Restart]] the failed element is dropped and
+ * the current extrapolation state is reset. For iterator-evaluation
failures, `Resume` and `Restart` both discard
+ * the corrupt iterator because it cannot be reused.
*
* '''Emits when''' downstream stops backpressuring, AND EITHER upstream
emits OR initial element is present OR
* `extrapolate` is non-empty and applicable
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
