You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "He-Pin (via GitHub)" <gi...@apache.org> on 2026/06/25 11:11:53 UTC
[PR] fix: add supervision strategy support for expand and extrapolate [pekko]
He-Pin opened a new pull request, #3185:
URL: https://github.com/apache/pekko/pull/3185
### Motivation
Issue #3110 tracks stream operators that should honor `SupervisionStrategy` for user functions.
`expand` called the user-provided expansion logic without supervision, so exceptions failed the stream unconditionally. Since `extrapolate` is implemented via `Expand`, the same gap applied there too.
### Modification
- Added supervision handling in `Expand` for:
- failures while creating extrapolation from an incoming element
- failures while evaluating iterator paths (`hasNext` / `next`)
- Supervision semantics:
- `Stop`: fail stage
- `Resume`: drop failed element and continue
- `Restart`: drop failed element, reset current extrapolation state, continue
- Updated Scala/Java API docs (`Flow`, `Source`, `SubFlow`, `SubSource`) to describe supervision behavior and clarify Resume vs Restart for `expand`.
- Updated operator docs:
- `expand.md`
- `extrapolate.md`
- Added directional regression tests:
- `FlowExpandSpec`: Stop/default Stop/Resume/Restart + iterator-evaluation failure paths
- `FlowExtrapolateSpec`: Stop/Resume/Restart for extrapolator failures during iterator evaluation
### Result
`expand` and `extrapolate` now honor `ActorAttributes.SupervisionStrategy` for user-function failures with explicit regression coverage for exceptional iterator paths.
which is now Apache licensed
### Tests
- `sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowExpandSpec org.apache.pekko.stream.scaladsl.FlowExtrapolateSpec"` (23/23 passed)
- `sbt "stream/mimaReportBinaryIssues"` (clean)
- `sbt "docs/paradox"` (passed)
### References
Refs #3110
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: add supervision strategy support for expand and extrapolate [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin merged PR #3185:
URL: https://github.com/apache/pekko/pull/3185
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: add supervision strategy support for expand and extrapolate [pekko]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3185:
URL: https://github.com/apache/pekko/pull/3185#discussion_r3487566344
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala:
##########
@@ -2501,12 +2501,16 @@ 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 upstream it will be backpressured by the downstream subscriber.
Review Comment:
The Scaladoc sentence compares "upstream" to itself ("faster than the upstream"), which is grammatically incorrect and unclear. It should compare upstream speed to downstream speed.
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala:
##########
@@ -3969,12 +3969,16 @@ 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 upstream it will be backpressured by the downstream subscriber.
Review Comment:
The Scaladoc sentence compares "upstream" to itself ("faster than the upstream"), which is grammatically incorrect and unclear. It should compare upstream speed to downstream speed.
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala:
##########
@@ -2566,12 +2566,16 @@ 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 upstream it will be backpressured by the downstream subscriber.
Review Comment:
The Scaladoc sentence compares "upstream" to itself ("faster than the upstream"), which is grammatically incorrect and unclear. It should compare upstream speed to downstream speed.
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala:
##########
@@ -1859,12 +1859,16 @@ 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 upstream it will be backpressured by the downstream subscriber.
Review Comment:
The Scaladoc sentence compares "upstream" to itself ("faster than the upstream"), which is grammatically incorrect and unclear. It should compare upstream speed to downstream speed.
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowExpandSpec.scala:
##########
@@ -150,6 +151,108 @@ 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() // Does it complete immediately?
Review Comment:
This inline test comment is phrased as a question, which makes the test intent look uncertain. Consider removing it or rephrasing as a statement.
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala:
##########
@@ -1891,12 +1891,16 @@ 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 upstream it will be backpressured by the downstream subscriber.
Review Comment:
The Scaladoc sentence compares "upstream" to itself ("faster than the upstream"), which is grammatically incorrect and unclear. It should compare upstream speed to downstream speed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org