Add opt-in stream error log throttling

stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/StageErrorLogThrottleSpec.scala

@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.impl.fusing
+
+import scala.concurrent.Await
+import scala.concurrent.duration._
+
+import org.apache.pekko
+import pekko.stream._
+import pekko.stream.impl.fusing.GraphStages.SimpleLinearGraphStage
+import pekko.stream.scaladsl._
+import pekko.stream.stage.GraphStageLogic
+import pekko.stream.stage.InHandler
+import pekko.stream.stage.OutHandler
+import pekko.stream.testkit.StreamSpec
+import pekko.testkit.EventFilter
+
+class StageErrorLogThrottleSpec extends StreamSpec("""
+    pekko.stream.materializer.stage-errors-log-throttle-period = 2s
+    pekko.loglevel = DEBUG
+  """) {
+
+  private def mkFailingStage: SimpleLinearGraphStage[Int] = new SimpleLinearGraphStage[Int] {
+    override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
+      new GraphStageLogic(shape) with InHandler with OutHandler {
+        setHandlers(in, out, this)
+        override def onPush(): Unit = throw new IllegalArgumentException("test failure")
+        override def onPull(): Unit = pull(in)
+      }
+  }
+
+  "Stage error log throttling" must {
+
+    "suppress repeated errors and flush count on stream finish" in {
+      // Use Broadcast to fan-out to 5 independent failing stages sharing one interpreter.
+      // Only the first error should be logged; the remaining 4 are suppressed by throttling.
+      EventFilter[IllegalArgumentException](pattern = "Error in stage.*", occurrences = 1).intercept {
+        val done = RunnableGraph
+          .fromGraph(GraphDSL.createGraph(Sink.ignore) { implicit b => sink =>
+            import GraphDSL.Implicits._
+            val bcast = b.add(Broadcast[Int](5, eagerCancel = false))
+            Source.single(1) ~> bcast
+            bcast.out(0)     ~> b.add(mkFailingStage) ~> sink
+            for (i <- 1 until 5) {
+              bcast.out(i) ~> b.add(mkFailingStage) ~> b.add(Sink.ignore)
+            }
+            ClosedShape
+          })
+          .run()
+        Await.ready(done, 3.seconds)
+      }
+    }
+
+    "always log a single error even with throttling enabled" in {
+      EventFilter[IllegalArgumentException](pattern = "Error in stage.*", occurrences = 1).intercept {
+        val result = Source.single(1).via(mkFailingStage).runWith(Sink.ignore)
+        Await.ready(result, 3.seconds)
+      }
+    }
+  }
+}
+
+class StageErrorLogThrottleDisabledSpec extends StreamSpec("""
+    pekko.stream.materializer.stage-errors-log-throttle-period = off
+  """) {
+
+  private def mkFailingStage: SimpleLinearGraphStage[Int] = new SimpleLinearGraphStage[Int] {
+    override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
+      new GraphStageLogic(shape) with InHandler with OutHandler {
+        setHandlers(in, out, this)
+        override def onPush(): Unit = throw new IllegalArgumentException("test failure")
+        override def onPull(): Unit = pull(in)
+      }
+  }
+
+  "Stage error log throttling when disabled" must {
+
+    "log every error individually without suppression" in {
+      // With throttling disabled, each stage error should be logged individually.
+      // Using 5 parallel failing stages via Broadcast, all 5 errors should appear.
+      EventFilter[IllegalArgumentException](pattern = "Error in stage.*", occurrences = 5).intercept {
+        val done = RunnableGraph
+          .fromGraph(GraphDSL.createGraph(Sink.ignore) { implicit b => sink =>
+            import GraphDSL.Implicits._
+            val bcast = b.add(Broadcast[Int](5, eagerCancel = false))
+            Source.single(1) ~> bcast
+            bcast.out(0)     ~> b.add(mkFailingStage) ~> sink
+            for (i <- 1 until 5) {
+              bcast.out(i) ~> b.add(mkFailingStage) ~> b.add(Sink.ignore)
+            }
+            ClosedShape
+          })
+          .run()
+        Await.ready(done, 3.seconds)
+      }
+    }
+
+    "log single error without suppression warning" in {
+      EventFilter[IllegalArgumentException](pattern = "Error in stage.*", occurrences = 1).intercept {
+        val result = Source.single(1).via(mkFailingStage).runWith(Sink.ignore)
+        Await.ready(result, 3.seconds)
+      }
+    }
+  }
+}

stream/src/main/resources/reference.conf

@@ -51,6 +51,12 @@ pekko {
         timeout = 5s
       }
 
+      # Period for throttling stage error log messages. When set, only the first
+      # error within this period is logged; subsequent errors are counted and
+      # a summary is logged when the period expires. Set to "off" to disable
+      # throttling (every error is logged individually).
+      stage-errors-log-throttle-period = off
+
       # Enable additional troubleshooting logging at DEBUG log level
       debug-logging = off
 

stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala

@@ -13,7 +13,7 @@
 
 package org.apache.pekko.stream.impl.fusing
 
-import java.util.concurrent.ThreadLocalRandom
+import java.util.concurrent.{ ThreadLocalRandom, TimeUnit }
 
 import scala.concurrent.Promise
 import scala.util.control.NonFatal
@@ -247,6 +247,26 @@ import pekko.stream.stage._
   private var _subFusingMaterializer: Materializer = _
   private lazy val defaultErrorReportingLogLevel = LogLevels.defaultErrorLevel(materializer.system)
 
+  // Throttle state for stage error logging
+  private[this] lazy val throttlePeriodNanos: Long = {
+    try {
+      val config = materializer.system.settings.config
+      val path = "pekko.stream.materializer.stage-errors-log-throttle-period"
+      val value = config.getString(path)
+      if (pekko.util.Helpers.toRootLowerCase(value) == "off") 0L
+      else {
+        val nanos = config.getDuration(path, TimeUnit.NANOSECONDS)
+        require(nanos > 0L, s"$path must be a positive duration or 'off', got: $value")
+        nanos
+      }
+    } catch {
+      case _: UnsupportedOperationException => 0L // NoMaterializer in tests
+    }
+  }
+  private[this] var lastErrorLogTime: Long = 0L
+  private[this] var errorLogInitialized: Boolean = false
+  private[this] var suppressedErrorCount: Int = 0
+
   def subFusingMaterializer: Materializer = _subFusingMaterializer
 
   // An event queue implemented as a circular buffer
@@ -339,6 +359,12 @@ import pekko.stream.stage._
    * Finalizes the state of all operators by calling postStop() (if necessary).
    */
   def finish(): Unit = {
+    // Flush any suppressed error count so operators know errors were throttled
+    if (suppressedErrorCount > 0) {
+      log.warning("{} additional stage error(s) were suppressed during throttle period",
+        suppressedErrorCount)
+      suppressedErrorCount = 0
+    }
     var i = 0
     while (i < logics.length) {
       val logic = logics(i)
@@ -382,16 +408,35 @@ import pekko.stream.stage._
         case Some(levels) => levels.onFailure
         case None         => defaultErrorReportingLogLevel
       }
-      logAt match {
-        case Logging.ErrorLevel =>
-          log.error(e, "Error in stage [{}]: {}", activeStage.toString, e.getMessage)
-        case Logging.WarningLevel =>
-          log.warning(e, "Error in stage [{}]: {}", activeStage.toString, e.getMessage)
-        case Logging.InfoLevel =>
-          log.info("Error in stage [{}]: {}", activeStage.toString, e.getMessage)
-        case Logging.DebugLevel =>
-          log.debug("Error in stage [{}]: {}", activeStage.toString, e.getMessage)
-        case _ => // Off, nop
+      val shouldLog = if (throttlePeriodNanos > 0L) {
+        val now = System.nanoTime()
+        if (!errorLogInitialized || now - lastErrorLogTime >= throttlePeriodNanos) {
+          if (suppressedErrorCount > 0) {
+            log.warning("{} additional stage error(s) were suppressed during throttle period",
+              suppressedErrorCount)
+          }
+          errorLogInitialized = true
+          lastErrorLogTime = now
+          suppressedErrorCount = 0
+          true
+        } else {
+          suppressedErrorCount += 1
+          false
+        }
+      } else true
+
+      if (shouldLog) {
+        logAt match {
+          case Logging.ErrorLevel =>
+            log.error(e, "Error in stage [{}]: {}", activeStage.toString, e.getMessage)
+          case Logging.WarningLevel =>
+            log.warning(e, "Error in stage [{}]: {}", activeStage.toString, e.getMessage)
+          case Logging.InfoLevel =>
+            log.info("Error in stage [{}]: {}", activeStage.toString, e.getMessage)
+          case Logging.DebugLevel =>
+            log.debug("Error in stage [{}]: {}", activeStage.toString, e.getMessage)
+          case _ => // Off, nop
+        }
       }
       activeStage.failStage(e)