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 0b9755a079 fix: add supervision strategy support for zipWith and
zipWithN zipper functions (#3186)
0b9755a079 is described below
commit 0b9755a079fe3c66e2bf1f8516ba613cb2410535
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Sun Jun 28 05:57:28 2026 +0800
fix: add supervision strategy support for zipWith and zipWithN zipper
functions (#3186)
* fix: add supervision strategy support for zipWith and zipWithN zipper
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 #3110
* fix: remove stale EventFilter from GraphZipWithSpec
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 #3110
---
.../stream/operators/Source-or-Flow/zipWith.md | 4 ++
.../paradox/stream/operators/Source/zipWithN.md | 5 ++-
.../pekko/stream/scaladsl/FlowZipWithSpec.scala | 41 ++++++++++++++++++--
.../pekko/stream/scaladsl/GraphZipWithNSpec.scala | 44 +++++++++++++++++++---
.../pekko/stream/scaladsl/GraphZipWithSpec.scala | 5 +--
.../stream/scaladsl/ZipWithApply.scala.template | 21 ++++++++++-
.../org/apache/pekko/stream/javadsl/Flow.scala | 6 +++
.../org/apache/pekko/stream/javadsl/Source.scala | 12 ++++++
.../org/apache/pekko/stream/javadsl/SubFlow.scala | 6 +++
.../apache/pekko/stream/javadsl/SubSource.scala | 6 +++
.../org/apache/pekko/stream/scaladsl/Flow.scala | 6 +++
.../org/apache/pekko/stream/scaladsl/Graph.scala | 29 +++++++++++++-
.../org/apache/pekko/stream/scaladsl/Source.scala | 6 +++
13 files changed, 175 insertions(+), 16 deletions(-)
diff --git a/docs/src/main/paradox/stream/operators/Source-or-Flow/zipWith.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/zipWith.md
index 0512ff1b22..1a54c919c5 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/zipWith.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/zipWith.md
@@ -15,6 +15,10 @@ Combines elements from multiple sources through a `combine`
function and passes
Combines elements from multiple sources through a `combine` function and
passes the
returned value downstream.
+This operator adheres to the `ActorAttributes.SupervisionStrategy` attribute
for exceptions thrown by the
+`combine` function. On `Supervision.Stop` the stream fails; on
`Supervision.Resume` and `Supervision.Restart`
+the failing zipped element is dropped and the stream continues.
+
See also:
* @ref:[zip](zip.md)
diff --git a/docs/src/main/paradox/stream/operators/Source/zipWithN.md
b/docs/src/main/paradox/stream/operators/Source/zipWithN.md
index 63c688fdc0..16527f3c4a 100644
--- a/docs/src/main/paradox/stream/operators/Source/zipWithN.md
+++ b/docs/src/main/paradox/stream/operators/Source/zipWithN.md
@@ -15,6 +15,10 @@ Combine the elements of multiple streams into a stream of
sequences using a comb
This operator is essentially the same as using @ref:[zipN](zipN.md) followed
by @ref[map](../Source-or-Flow/map.md)
to turn the zipped sequence into an arbitrary object to emit downstream.
+This operator adheres to the `ActorAttributes.SupervisionStrategy` attribute
for exceptions thrown by the `zipper`
+function. On `Supervision.Stop` the stream fails; on `Supervision.Resume` and
`Supervision.Restart` the failing zipped
+element is dropped and the stream continues.
+
See also:
* @ref:[zipN](zipN.md)
@@ -48,4 +52,3 @@ Note how it stops as soon as any of the original sources
reaches its end.
@@@
-
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowZipWithSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowZipWithSpec.scala
index cb4827d126..526d619c9d 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowZipWithSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowZipWithSpec.scala
@@ -14,11 +14,13 @@
package org.apache.pekko.stream.scaladsl
import org.apache.pekko
+import pekko.stream.ActorAttributes
+import pekko.stream.Supervision
import pekko.stream.testkit.{ BaseTwoStreamsSetup, TestSubscriber }
import org.reactivestreams.Publisher
+import scala.concurrent.Await
import scala.concurrent.duration._
-import pekko.testkit.EventFilter
import scala.annotation.nowarn
@nowarn // keep unused imports
@@ -68,15 +70,46 @@ class FlowZipWithSpec extends BaseTwoStreamsSetup {
probe.expectNext(1 / -2)
probe.expectNext(2 / -1)
- EventFilter[ArithmeticException](occurrences = 1).intercept {
- subscription.request(2)
- }
+ subscription.request(2)
probe.expectError() match {
case a: java.lang.ArithmeticException => a.getMessage should be("/ by
zero")
}
probe.expectNoMessage(200.millis)
}
+ "fail stream when zipper throws and supervision is Stop" in {
+ val ex = new RuntimeException("boom")
+ val result = Source(1 to 4)
+ .zipWith(Source(1 to 4))((a, b) => if (a == 3) throw ex else a + b)
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.stoppingDecider))
+ .runWith(Sink.seq)
+ result.failed.futureValue shouldBe ex
+ }
+
+ "fail stream when zipper throws and supervision defaults to Stop" in {
+ val ex = new RuntimeException("boom")
+ val result = Source(1 to 4)
+ .zipWith(Source(1 to 4))((a, b) => if (a == 3) throw ex else a + b)
+ .runWith(Sink.seq)
+ result.failed.futureValue shouldBe ex
+ }
+
+ "resume when zipper throws and drop failed zipped element" in {
+ val future = Source(1 to 4)
+ .zipWith(Source(1 to 4))((a, b) => if (a == 3) throw new
RuntimeException("boom") else a + b)
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
+ .runWith(Sink.seq)
+ Await.result(future, 3.seconds) shouldBe Seq(2, 4, 8)
+ }
+
+ "restart when zipper throws and drop failed zipped element" in {
+ val future = Source(1 to 4)
+ .zipWith(Source(1 to 4))((a, b) => if (a == 3) throw new
RuntimeException("boom") else a + b)
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.restartingDecider))
+ .runWith(Sink.seq)
+ Await.result(future, 3.seconds) shouldBe Seq(2, 4, 8)
+ }
+
commonTests()
"work with one immediately completed and one nonempty publisher" in {
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/GraphZipWithNSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/GraphZipWithNSpec.scala
index deacff1f2b..260456aba1 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/GraphZipWithNSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/GraphZipWithNSpec.scala
@@ -14,12 +14,13 @@
package org.apache.pekko.stream.scaladsl
import scala.collection.immutable
+import scala.concurrent.Await
import scala.concurrent.duration._
import org.apache.pekko
import pekko.stream._
-import pekko.stream.testkit._
-import pekko.testkit.EventFilter
+import pekko.stream.testkit.TestSubscriber
+import pekko.stream.testkit.TwoStreamsSetup
class GraphZipWithNSpec extends TwoStreamsSetup {
import GraphDSL.Implicits._
@@ -86,9 +87,7 @@ class GraphZipWithNSpec extends TwoStreamsSetup {
probe.expectNext(1 / 1 / -2)
probe.expectNext(1 / 2 / -1)
- EventFilter[ArithmeticException](occurrences = 1).intercept {
- subscription.request(2)
- }
+ subscription.request(2)
probe.expectError() match {
case a: java.lang.ArithmeticException => a.getMessage should be("/ by
zero")
case unexpected => throw new
RuntimeException(s"Unexpected: $unexpected")
@@ -96,6 +95,41 @@ class GraphZipWithNSpec extends TwoStreamsSetup {
probe.expectNoMessage(200.millis)
}
+ "fail stream when zipper throws and supervision is Stop" in {
+ val ex = new RuntimeException("boom")
+ val result = Source
+ .zipWithN[Int, Int](s => if (s.head == 3) throw ex else
s.sum)(immutable.Seq(Source(1 to 4), Source(1 to 4)))
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.stoppingDecider))
+ .runWith(Sink.seq)
+ result.failed.futureValue shouldBe ex
+ }
+
+ "fail stream when zipper throws and supervision defaults to Stop" in {
+ val ex = new RuntimeException("boom")
+ val result = Source
+ .zipWithN[Int, Int](s => if (s.head == 3) throw ex else
s.sum)(immutable.Seq(Source(1 to 4), Source(1 to 4)))
+ .runWith(Sink.seq)
+ result.failed.futureValue shouldBe ex
+ }
+
+ "resume when zipper throws and drop failed zipped element" in {
+ val future = Source
+ .zipWithN[Int, Int](s => if (s.head == 3) throw new
RuntimeException("boom") else s.sum)(
+ immutable.Seq(Source(1 to 4), Source(1 to 4)))
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
+ .runWith(Sink.seq)
+ Await.result(future, 3.seconds) shouldBe Seq(2, 4, 8)
+ }
+
+ "restart when zipper throws and drop failed zipped element" in {
+ val future = Source
+ .zipWithN[Int, Int](s => if (s.head == 3) throw new
RuntimeException("boom") else s.sum)(
+ immutable.Seq(Source(1 to 4), Source(1 to 4)))
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.restartingDecider))
+ .runWith(Sink.seq)
+ Await.result(future, 3.seconds) shouldBe Seq(2, 4, 8)
+ }
+
commonTests()
"work with one immediately completed and one nonempty publisher" in {
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/GraphZipWithSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/GraphZipWithSpec.scala
index b2a9d69856..5428dce8c7 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/GraphZipWithSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/GraphZipWithSpec.scala
@@ -18,7 +18,6 @@ import scala.concurrent.duration._
import org.apache.pekko
import pekko.stream._
import pekko.stream.testkit._
-import pekko.testkit.EventFilter
class GraphZipWithSpec extends TwoStreamsSetup {
import GraphDSL.Implicits._
@@ -85,9 +84,7 @@ class GraphZipWithSpec extends TwoStreamsSetup {
probe.expectNext(1 / -2)
probe.expectNext(2 / -1)
- EventFilter[ArithmeticException](occurrences = 1).intercept {
- subscription.request(2)
- }
+ subscription.request(2)
probe.expectError() match {
case a: java.lang.ArithmeticException => a.getMessage should be("/ by
zero")
case unexpected => throw new
RuntimeException(s"Unexpected: $unexpected")
diff --git
a/stream/src/main/boilerplate/org/apache/pekko/stream/scaladsl/ZipWithApply.scala.template
b/stream/src/main/boilerplate/org/apache/pekko/stream/scaladsl/ZipWithApply.scala.template
index a1239db97a..2cdcdf5aaa 100644
---
a/stream/src/main/boilerplate/org/apache/pekko/stream/scaladsl/ZipWithApply.scala.template
+++
b/stream/src/main/boilerplate/org/apache/pekko/stream/scaladsl/ZipWithApply.scala.template
@@ -13,7 +13,10 @@
package org.apache.pekko.stream.scaladsl
+import scala.util.control.NonFatal
+
import org.apache.pekko.stream._
+import org.apache.pekko.stream.ActorAttributes.SupervisionStrategy
import org.apache.pekko.stream.impl.ContextPropagation
import org.apache.pekko.stream.stage._
@@ -41,6 +44,7 @@ class ZipWith1[[#A1#], O] (val zipper: ([#A1#]) => O) extends
GraphStage[FanInSh
]
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
new GraphStageLogic(shape) {
+ private lazy val decider =
inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider
var pending = ##0
// Without this field the completion signalling would take one extra pull
var willShutDown = false
@@ -48,7 +52,22 @@ class ZipWith1[[#A1#], O] (val zipper: ([#A1#]) => O)
extends GraphStage[FanInSh
private def pushAll(): Unit = {
contextPropagation.resumeContext()
- push(out, zipper([#grab(in0)#]))
+ try push(out, zipper([#grab(in0)#]))
+ catch {
+ case NonFatal(ex) =>
+ decider(ex) match {
+ case Supervision.Stop =>
+ failStage(ex)
+ case Supervision.Resume | Supervision.Restart =>
+ if (willShutDown) completeStage()
+ else {
+ pending += shape.inlets.size
+ [#pull(in0)#
+ ]
+ }
+ }
+ return
+ }
if (willShutDown) completeStage()
else {
[#pull(in0)#
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 3015f02e31..91fea73f05 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
@@ -3847,6 +3847,12 @@ final class Flow[In, Out, Mat](delegate:
scaladsl.Flow[In, Out, Mat]) extends Gr
* Put together the elements of current [[Flow]] and the given [[Source]]
* into a stream of combined elements using a combiner function.
*
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the combiner function throws and the supervision decision is
[[pekko.stream.Supervision#stop]]
+ * the stream fails. If the supervision decision is
[[pekko.stream.Supervision#resume]] or
+ * [[pekko.stream.Supervision#restart]] the zipped element is dropped and
the stream continues.
+ *
* '''Emits when''' all of the inputs have an element available
*
* '''Backpressures when''' downstream backpressures
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 479807d37f..5ccd8a3870 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
@@ -589,6 +589,12 @@ object Source {
/**
* Combine the elements of multiple streams into a stream of lists using a
combiner function.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the combiner function throws and the supervision decision is
[[pekko.stream.Supervision#stop]]
+ * the stream fails. If the supervision decision is
[[pekko.stream.Supervision#resume]] or
+ * [[pekko.stream.Supervision#restart]] the zipped element is dropped and
the stream continues.
*/
def zipWithN[T, O](
zipper: function.Function[java.util.List[T], O],
@@ -1936,6 +1942,12 @@ final class Source[Out, Mat](delegate:
scaladsl.Source[Out, Mat]) extends Graph[
* Put together the elements of current [[Source]] and the given one
* into a stream of combined elements using a combiner function.
*
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the combiner function throws and the supervision decision is
[[pekko.stream.Supervision#stop]]
+ * the stream fails. If the supervision decision is
[[pekko.stream.Supervision#resume]] or
+ * [[pekko.stream.Supervision#restart]] the zipped element is dropped and
the stream continues.
+ *
* '''Emits when''' all of the inputs has an element available
*
* '''Backpressures when''' downstream backpressures
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 ca1651bc69..00e81790f8 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
@@ -2570,6 +2570,12 @@ final class SubFlow[In, Out, Mat](
* Put together the elements of current [[Flow]] and the given [[Source]]
* into a stream of combined elements using a combiner function.
*
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the combiner function throws and the supervision decision is
[[pekko.stream.Supervision#stop]]
+ * the stream fails. If the supervision decision is
[[pekko.stream.Supervision#resume]] or
+ * [[pekko.stream.Supervision#restart]] the zipped element is dropped and
the stream continues.
+ *
* '''Emits when''' all of the inputs has an element available
*
* '''Backpressures when''' downstream backpressures
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 dc17efe2f1..bf55e25495 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
@@ -2537,6 +2537,12 @@ final class SubSource[Out, Mat](
* Put together the elements of current [[Flow]] and the given [[Source]]
* into a stream of combined elements using a combiner function.
*
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the combiner function throws and the supervision decision is
[[pekko.stream.Supervision#stop]]
+ * the stream fails. If the supervision decision is
[[pekko.stream.Supervision#resume]] or
+ * [[pekko.stream.Supervision#restart]] the zipped element is dropped and
the stream continues.
+ *
* '''Emits when''' all of the inputs has an element available
*
* '''Backpressures when''' downstream backpressures
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 c35fa5dfb6..2a4657e5c8 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
@@ -3420,6 +3420,12 @@ trait FlowOps[+Out, +Mat] {
* Put together the elements of current flow and the given [[Source]]
* into a stream of combined elements using a combiner function.
*
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the combiner function throws and the supervision decision is
[[pekko.stream.Supervision.Stop]]
+ * the stream fails. If the supervision decision is
[[pekko.stream.Supervision.Resume]] or
+ * [[pekko.stream.Supervision.Restart]] the zipped element is dropped and
the stream continues.
+ *
* '''Emits when''' all of the inputs have an element available
*
* '''Backpressures when''' downstream backpressures
diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Graph.scala
b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Graph.scala
index 6cde3ffa9f..415af7ea12 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Graph.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Graph.scala
@@ -1096,6 +1096,12 @@ final class ZipLatest[A, B](eagerComplete: Boolean)
extends ZipLatestWith2[A, B,
* '''Completes when''' any upstream completes
*
* '''Cancels when''' downstream cancels
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the combiner function throws and the supervision decision is
[[pekko.stream.Supervision.Stop]]
+ * the stream fails. If the supervision decision is
[[pekko.stream.Supervision.Resume]] or
+ * [[pekko.stream.Supervision.Restart]] the zipped element is dropped and the
stream continues.
*/
object ZipWith extends ZipWithApply
@@ -1219,6 +1225,12 @@ object ZipWithN {
* '''Completes when''' any upstream completes
*
* '''Cancels when''' downstream cancels
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the combiner function throws and the supervision decision is
[[pekko.stream.Supervision.Stop]]
+ * the stream fails. If the supervision decision is
[[pekko.stream.Supervision.Resume]] or
+ * [[pekko.stream.Supervision.Restart]] the zipped element is dropped and the
stream continues.
*/
class ZipWithN[A, O](zipper: immutable.Seq[A] => O)(n: Int) extends
GraphStage[UniformFanInShape[A, O]] {
override def initialAttributes = DefaultAttributes.zipWithN
@@ -1227,6 +1239,7 @@ class ZipWithN[A, O](zipper: immutable.Seq[A] => O)(n:
Int) extends GraphStage[U
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
new GraphStageLogic(shape) with OutHandler {
+ private lazy val decider =
inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider
var pending = 0
// Without this field the completion signalling would take one extra pull
var willShutDown = false
@@ -1238,7 +1251,21 @@ class ZipWithN[A, O](zipper: immutable.Seq[A] => O)(n:
Int) extends GraphStage[U
private def pushAll(): Unit = {
contextPropagation.resumeContext()
- push(out, zipper(shape.inlets.map(grabInlet)))
+ try push(out, zipper(shape.inlets.map(grabInlet)))
+ catch {
+ case NonFatal(ex) =>
+ decider(ex) match {
+ case Supervision.Stop =>
+ failStage(ex)
+ case Supervision.Resume | Supervision.Restart =>
+ if (willShutDown) completeStage()
+ else {
+ pending += n
+ shape.inlets.foreach(pullInlet)
+ }
+ }
+ return
+ }
if (willShutDown) completeStage()
else shape.inlets.foreach(pullInlet)
}
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala
b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala
index 4cd4594e1d..8b4e60d539 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala
@@ -887,6 +887,12 @@ object Source {
/**
* Combine the elements of multiple streams into a stream of sequences using
a combiner function.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * If the combiner function throws and the supervision decision is
[[pekko.stream.Supervision.Stop]]
+ * the stream fails. If the supervision decision is
[[pekko.stream.Supervision.Resume]] or
+ * [[pekko.stream.Supervision.Restart]] the zipped element is dropped and
the stream continues.
*/
def zipWithN[T, O](zipper: immutable.Seq[T] => O)(sources:
immutable.Seq[Source[T, ?]]): Source[O, NotUsed] = {
val source = sources match {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
