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 f57cc814c7 fix(stream): log supervision Resume/Restart in MapAsync
operators (#3124)
f57cc814c7 is described below
commit f57cc814c7d475fd58743e0ad8c6592c5aaf35b4
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Tue Jun 23 01:25:12 2026 +0800
fix(stream): log supervision Resume/Restart in MapAsync operators (#3124)
Motivation:
MapAsync, MapAsyncUnordered, and MapAsyncPartitioned silently skip
failed elements when the supervision strategy returns Resume or
Restart, making production debugging difficult.
Modification:
Add StageLogging mixin and log supervision failures at the configured
error level (respecting stage-errors-default-log-level and per-stage
LogLevels.onFailure attribute) before skipping elements. Logging is
placed in pushNextIfPossible / futureCompleted / onPush sync handler
to avoid double-logging via Holder's directive cache.
Result:
Supervision Resume/Restart decisions now produce log entries with
exception details, enabling users to diagnose dropped elements.
Tests:
- FlowMapAsyncSpec: 3 new log verification tests (async, sync, Off)
- FlowMapAsyncUnorderedSpec: 3 new log verification tests
- FlowMapAsyncPartitionedSpec: 3 new log verification tests (ordered +
unordered + Off)
- All 74 existing + new tests pass
References:
Fixes #3103
---
.../scaladsl/FlowMapAsyncPartitionedSpec.scala | 55 +++++++++++++++++
.../pekko/stream/scaladsl/FlowMapAsyncSpec.scala | 47 ++++++++++++++
.../scaladsl/FlowMapAsyncUnorderedSpec.scala | 53 ++++++++++++++++
.../apache/pekko/stream/MapAsyncPartitioned.scala | 41 ++++++++++--
.../org/apache/pekko/stream/impl/fusing/Ops.scala | 72 ++++++++++++++++++++--
5 files changed, 258 insertions(+), 10 deletions(-)
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapAsyncPartitionedSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapAsyncPartitionedSpec.scala
index 8b3635050d..1920af03c2 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapAsyncPartitionedSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapAsyncPartitionedSpec.scala
@@ -27,10 +27,12 @@ import scala.concurrent.duration._
import org.apache.pekko
import pekko.Done
import pekko.stream.ActorAttributes
+import pekko.stream.Attributes
import pekko.stream.Supervision
import pekko.stream.testkit._
import pekko.stream.testkit.scaladsl.TestSink
import pekko.stream.testkit.scaladsl.TestSource
+import pekko.testkit.EventFilter
import pekko.testkit.TestDuration
import pekko.testkit.TestLatch
import pekko.testkit.TestProbe
@@ -578,5 +580,58 @@ class FlowMapAsyncPartitionedSpec extends StreamSpec with
WithLogCapturing {
else Future.successful(elem)
}
}
+
+ "log error when future fails with resume supervision (ordered)" in {
+ EventFilter.error(
+ start = "Supervision strategy resumed after exception in
MapAsyncPartitioned",
+ occurrences = 1).intercept {
+ val result = Source(1 to 5)
+ .mapAsyncPartitioned(4)(_ % 2) { (n, _) =>
+ if (n == 3) Future.failed(TE("err3"))
+ else Future.successful(n)
+ }
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
+ .runWith(Sink.seq)
+
+ result.futureValue should ===(Seq(1, 2, 4, 5))
+ }
+ }
+
+ "log error when future fails with resume supervision (unordered)" in {
+ EventFilter.error(
+ start = "Supervision strategy resumed after exception in
MapAsyncPartitioned",
+ occurrences = 1).intercept {
+ val result = Source(1 to 5)
+ .mapAsyncPartitionedUnordered(4)(_ % 2) { (n, _) =>
+ if (n == 3) Future.failed(TE("err3"))
+ else Future.successful(n)
+ }
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
+ .runWith(Sink.seq)
+
+ result.futureValue.toSet should ===(Set(1, 2, 4, 5))
+ }
+ }
+
+ "not log when log level is Off and resume supervision is used" in {
+ EventFilter.error(
+ start = "Supervision strategy resumed after exception in
MapAsyncPartitioned",
+ occurrences = 0).intercept {
+ val result = Source(1 to 5)
+ .mapAsyncPartitioned(4)(_ % 2) { (n, _) =>
+ if (n == 3) Future.failed(TE("err3"))
+ else Future.successful(n)
+ }
+ .withAttributes(
+ ActorAttributes.supervisionStrategy(Supervision.resumingDecider)
and
+ Attributes.logLevels(
+ onElement = Attributes.LogLevels.Off,
+ onFinish = Attributes.LogLevels.Off,
+ onFailure = Attributes.LogLevels.Off))
+ .runWith(Sink.seq)
+
+ result.futureValue should ===(Seq(1, 2, 4, 5))
+ }
+ }
}
}
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapAsyncSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapAsyncSpec.scala
index 281b0c3560..37ad2fb353 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapAsyncSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapAsyncSpec.scala
@@ -28,11 +28,13 @@ import scala.util.control.NoStackTrace
import org.apache.pekko
import pekko.stream.ActorAttributes
import pekko.stream.ActorAttributes.supervisionStrategy
+import pekko.stream.Attributes
import pekko.stream.Supervision
import pekko.stream.Supervision.resumingDecider
import pekko.stream.testkit._
import pekko.stream.testkit.Utils._
import pekko.stream.testkit.scaladsl.TestSink
+import pekko.testkit.EventFilter
import pekko.testkit.TestDuration
import pekko.testkit.TestLatch
import pekko.testkit.TestProbe
@@ -552,6 +554,51 @@ class FlowMapAsyncSpec extends StreamSpec {
failCount.get() should ===(1)
}
+ "log error when future fails with resume supervision" in {
+ EventFilter.error(start = "Supervision strategy resumed after exception
in MapAsync", occurrences = 1).intercept {
+ val result = Source(1 to 5)
+ .mapAsync(4)(n =>
+ if (n == 3) Future.failed(new TE("err3"))
+ else Future.successful(n))
+ .withAttributes(supervisionStrategy(resumingDecider))
+ .runWith(Sink.seq)
+
+ result.futureValue should ===(Seq(1, 2, 4, 5))
+ }
+ }
+
+ "log error when mapAsync function throws with resume supervision" in {
+ implicit val ec = system.dispatcher
+ EventFilter.error(start = "Supervision strategy resumed after exception
in MapAsync", occurrences = 1).intercept {
+ val result = Source(1 to 5)
+ .mapAsync(4)(n =>
+ if (n == 3) throw TE("err-sync")
+ else Future(n))
+ .withAttributes(supervisionStrategy(resumingDecider))
+ .runWith(Sink.seq)
+
+ result.futureValue should ===(Seq(1, 2, 4, 5))
+ }
+ }
+
+ "not log when log level is Off and resume supervision is used" in {
+ EventFilter.error(start = "Supervision strategy resumed after exception
in MapAsync", occurrences = 0).intercept {
+ val result = Source(1 to 5)
+ .mapAsync(4)(n =>
+ if (n == 3) Future.failed(new TE("err3"))
+ else Future.successful(n))
+ .withAttributes(
+ supervisionStrategy(resumingDecider) and
+ Attributes.logLevels(
+ onElement = Attributes.LogLevels.Off,
+ onFinish = Attributes.LogLevels.Off,
+ onFailure = Attributes.LogLevels.Off))
+ .runWith(Sink.seq)
+
+ result.futureValue should ===(Seq(1, 2, 4, 5))
+ }
+ }
+
}
}
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapAsyncUnorderedSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapAsyncUnorderedSpec.scala
index 21d0a4a566..8c7f5ca595 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapAsyncUnorderedSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapAsyncUnorderedSpec.scala
@@ -26,9 +26,11 @@ import scala.util.control.NoStackTrace
import org.apache.pekko
import pekko.stream.ActorAttributes.supervisionStrategy
+import pekko.stream.Attributes
import pekko.stream.Supervision.resumingDecider
import pekko.stream.testkit._
import pekko.stream.testkit.scaladsl._
+import pekko.testkit.EventFilter
import pekko.testkit.TestDuration
import pekko.testkit.TestLatch
import pekko.testkit.TestProbe
@@ -376,5 +378,56 @@ class FlowMapAsyncUnorderedSpec extends StreamSpec {
}
}
+ "log error when future fails with resume supervision" in {
+ EventFilter.error(
+ start = "Supervision strategy resumed after exception in
MapAsyncUnordered",
+ occurrences = 1).intercept {
+ val result = Source(1 to 5)
+ .mapAsyncUnordered(4)(n =>
+ if (n == 3) Future.failed(new Exception("err3") with NoStackTrace)
+ else Future.successful(n))
+ .withAttributes(supervisionStrategy(resumingDecider))
+ .runWith(Sink.seq)
+
+ result.futureValue.toSet should ===(Set(1, 2, 4, 5))
+ }
+ }
+
+ "log error when mapAsyncUnordered function throws with resume supervision"
in {
+ implicit val ec = system.dispatcher
+ EventFilter.error(
+ start = "Supervision strategy resumed after exception in
MapAsyncUnordered",
+ occurrences = 1).intercept {
+ val result = Source(1 to 5)
+ .mapAsyncUnordered(4)(n =>
+ if (n == 3) throw Utils.TE("err-sync")
+ else Future(n))
+ .withAttributes(supervisionStrategy(resumingDecider))
+ .runWith(Sink.seq)
+
+ result.futureValue.toSet should ===(Set(1, 2, 4, 5))
+ }
+ }
+
+ "not log when log level is Off and resume supervision is used" in {
+ EventFilter.error(
+ start = "Supervision strategy resumed after exception in
MapAsyncUnordered",
+ occurrences = 0).intercept {
+ val result = Source(1 to 5)
+ .mapAsyncUnordered(4)(n =>
+ if (n == 3) Future.failed(new Exception("err3") with NoStackTrace)
+ else Future.successful(n))
+ .withAttributes(
+ supervisionStrategy(resumingDecider) and
+ Attributes.logLevels(
+ onElement = Attributes.LogLevels.Off,
+ onFinish = Attributes.LogLevels.Off,
+ onFailure = Attributes.LogLevels.Off))
+ .runWith(Sink.seq)
+
+ result.futureValue.toSet should ===(Set(1, 2, 4, 5))
+ }
+ }
+
}
}
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/MapAsyncPartitioned.scala
b/stream/src/main/scala/org/apache/pekko/stream/MapAsyncPartitioned.scala
index 565d8a2b51..b7ca9219d2 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/MapAsyncPartitioned.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/MapAsyncPartitioned.scala
@@ -25,7 +25,10 @@ import scala.util.control.{ NoStackTrace, NonFatal }
import org.apache.pekko
import pekko.annotation.InternalApi
+import pekko.event.Logging
+import pekko.event.Logging.LogLevel
import pekko.stream.ActorAttributes.SupervisionStrategy
+import pekko.stream.Attributes.LogLevels
import pekko.stream.stage._
import pekko.util.OptionVal
@@ -82,9 +85,11 @@ private[stream] final class MapAsyncPartitioned[In, Out,
Partition](
override val shape: FlowShape[In, Out] = FlowShape(in, out)
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
- new GraphStageLogic(shape) with InHandler with OutHandler {
+ new GraphStageLogic(shape) with InHandler with OutHandler with
StageLogging {
private val contextPropagation = pekko.stream.impl.ContextPropagation()
+ override protected def logSource: Class[?] =
classOf[MapAsyncPartitioned[?, ?, ?]]
+
private final class Contextual[T](context: AnyRef, val element: T) {
private var suspended = false
@@ -107,6 +112,30 @@ private[stream] final class MapAsyncPartitioned[In, Out,
Partition](
private var partitionsInProgress: mutable.Set[Partition] = _
private var buffer: mutable.Queue[(Partition, Contextual[Holder[In,
Out]])] = _
+ private def logSupervisionFailure(ex: Throwable): Unit = {
+ val logAt: LogLevel = inheritedAttributes.get[LogLevels] match {
+ case Some(levels) => levels.onFailure
+ case None => LogLevels.defaultErrorLevel(materializer.system)
+ }
+ logAt match {
+ case Logging.ErrorLevel =>
+ log.error(ex, "Supervision strategy resumed after exception in
MapAsyncPartitioned: {}", ex.getMessage)
+ case Logging.WarningLevel =>
+ log.warning(ex, "Supervision strategy resumed after exception in
MapAsyncPartitioned: {}", ex.getMessage)
+ case Logging.InfoLevel =>
+ log.info(
+ "Supervision strategy resumed after exception in
MapAsyncPartitioned: {}: {}",
+ Logging.simpleName(ex.getClass),
+ ex.getMessage)
+ case Logging.DebugLevel =>
+ log.debug(
+ "Supervision strategy resumed after exception in
MapAsyncPartitioned: {}: {}",
+ Logging.simpleName(ex.getClass),
+ ex.getMessage)
+ case _ => // Off, nop
+ }
+ }
+
private val futureCB = getAsyncCallback[Holder[In, Out]](holder =>
holder.out match {
case Success(_) => pushNextIfPossible()
@@ -142,7 +171,9 @@ private[stream] final class MapAsyncPartitioned[In, Out,
Partition](
wrappedInput.suspend()
}
} catch {
- case NonFatal(ex) => if (decider(ex) == Supervision.Stop)
failStage(ex)
+ case NonFatal(ex) =>
+ if (decider(ex) == Supervision.Stop) failStage(ex)
+ else logSupervisionFailure(ex)
}
pullIfNeeded()
@@ -201,7 +232,8 @@ private[stream] final class MapAsyncPartitioned[In, Out,
Partition](
case Supervision.Stop =>
failStage(ex)
case _ =>
- // try next element
+ // try next element
+ logSupervisionFailure(ex)
}
case Failure(ex) =>
// fatal exception in buffer, not sure that it can actually
happen, but for good measure
@@ -242,7 +274,8 @@ private[stream] final class MapAsyncPartitioned[In, Out,
Partition](
case Supervision.Stop =>
failStage(ex)
case _ =>
- // try next element
+ // try next element
+ logSupervisionFailure(ex)
}
case Failure(ex) =>
// fatal exception in buffer, not sure that it can actually
happen, but for good measure
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 3832ccf269..010260e03d 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
@@ -1296,11 +1296,37 @@ private[stream] object Collect {
override val shape = FlowShape(in, out)
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
- new GraphStageLogic(shape) with InHandler with OutHandler {
+ new GraphStageLogic(shape) with InHandler with OutHandler with
StageLogging {
+
+ override protected def logSource: Class[?] = classOf[MapAsync[?, ?]]
lazy val decider =
inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider
var buffer: BufferImpl[Holder[Out]] = _
+ private def logSupervisionFailure(ex: Throwable): Unit = {
+ val logAt: LogLevel = inheritedAttributes.get[LogLevels] match {
+ case Some(levels) => levels.onFailure
+ case None => LogLevels.defaultErrorLevel(materializer.system)
+ }
+ logAt match {
+ case Logging.ErrorLevel =>
+ log.error(ex, "Supervision strategy resumed after exception in
MapAsync: {}", ex.getMessage)
+ case Logging.WarningLevel =>
+ log.warning(ex, "Supervision strategy resumed after exception in
MapAsync: {}", ex.getMessage)
+ case Logging.InfoLevel =>
+ log.info(
+ "Supervision strategy resumed after exception in MapAsync: {}:
{}",
+ Logging.simpleName(ex.getClass),
+ ex.getMessage)
+ case Logging.DebugLevel =>
+ log.debug(
+ "Supervision strategy resumed after exception in MapAsync: {}:
{}",
+ Logging.simpleName(ex.getClass),
+ ex.getMessage)
+ case _ => // Off, nop
+ }
+ }
+
private val futureCB = getAsyncCallback[Holder[Out]](holder =>
holder.elem match {
case Success(_) => pushNextIfPossible()
@@ -1348,7 +1374,9 @@ private[stream] object Collect {
} catch {
// this logic must only be executed if f throws, not if the future
is failed
- case NonFatal(ex) => if (decider(ex) == Supervision.Stop)
failStage(ex)
+ case NonFatal(ex) =>
+ if (decider(ex) == Supervision.Stop) failStage(ex)
+ else logSupervisionFailure(ex)
}
pullIfNeeded()
@@ -1380,6 +1408,7 @@ private[stream] object Collect {
case Supervision.Stop => failStage(ex)
case _ =>
// try next element
+ logSupervisionFailure(ex)
pushNextIfPossible()
}
case Failure(ex) =>
@@ -1413,9 +1442,11 @@ private[stream] object Collect {
override val shape = FlowShape(in, out)
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
- new GraphStageLogic(shape) with InHandler with OutHandler {
+ new GraphStageLogic(shape) with InHandler with OutHandler with
StageLogging {
override def toString = s"MapAsyncUnordered.Logic(inFlight=$inFlight,
buffer=$buffer)"
+ override protected def logSource: Class[?] =
classOf[MapAsyncUnordered[?, ?]]
+
private lazy val decider =
inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider
private var inFlight = 0
@@ -1424,6 +1455,30 @@ private[stream] object Collect {
private def todo: Int = inFlight + buffer.used
+ private def logSupervisionFailure(ex: Throwable): Unit = {
+ val logAt: LogLevel = inheritedAttributes.get[LogLevels] match {
+ case Some(levels) => levels.onFailure
+ case None => LogLevels.defaultErrorLevel(materializer.system)
+ }
+ logAt match {
+ case Logging.ErrorLevel =>
+ log.error(ex, "Supervision strategy resumed after exception in
MapAsyncUnordered: {}", ex.getMessage)
+ case Logging.WarningLevel =>
+ log.warning(ex, "Supervision strategy resumed after exception in
MapAsyncUnordered: {}", ex.getMessage)
+ case Logging.InfoLevel =>
+ log.info(
+ "Supervision strategy resumed after exception in
MapAsyncUnordered: {}: {}",
+ Logging.simpleName(ex.getClass),
+ ex.getMessage)
+ case Logging.DebugLevel =>
+ log.debug(
+ "Supervision strategy resumed after exception in
MapAsyncUnordered: {}: {}",
+ Logging.simpleName(ex.getClass),
+ ex.getMessage)
+ case _ => // Off, nop
+ }
+ }
+
override def preStart(): Unit = buffer = BufferImpl(parallelism,
inheritedAttributes)
def futureCompleted(result: Try[Out]): Unit = {
@@ -1441,8 +1496,11 @@ private[stream] object Collect {
else if (!hasBeenPulled(in)) tryPull(in)
case Failure(ex) =>
if (decider(ex) == Supervision.Stop) failStage(ex)
- else if (isCompleted) completeStage()
- else if (!hasBeenPulled(in)) tryPull(in)
+ else {
+ logSupervisionFailure(ex)
+ if (isCompleted) completeStage()
+ else if (!hasBeenPulled(in)) tryPull(in)
+ }
}
}
@@ -1455,7 +1513,9 @@ private[stream] object Collect {
case Some(v) => futureCompleted(v)
}
} catch {
- case NonFatal(ex) => if (decider(ex) == Supervision.Stop)
failStage(ex)
+ case NonFatal(ex) =>
+ if (decider(ex) == Supervision.Stop) failStage(ex)
+ else logSupervisionFailure(ex)
}
if (todo < parallelism && !hasBeenPulled(in)) tryPull(in)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
