This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch fix/mapasync-supervision-logging-3103 in repository https://gitbox.apache.org/repos/asf/pekko.git
commit cede5e633c2a07f5435af4a36f07b5cdaf01da8e Author: 虎鸣 <[email protected]> AuthorDate: Tue Jun 23 00:23:07 2026 +0800 fix(stream): log supervision Resume/Restart in MapAsync operators 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]
