Allow specifying a log level when logging exceptions being seen the SupervisorStrategy (#27147)

* Allow specifying a log level when logging exceptions being seen the SupervisorStrategy. #27133
* Add tests which verify Supervisor logging behaviour for provided log level

akka-actor-typed-tests/src/test/scala/akka/actor/typed/SupervisionSpec.scala

@@ -16,10 +16,11 @@ import akka.testkit.EventFilter
 import akka.actor.testkit.typed.scaladsl._
 import akka.actor.testkit.typed._
 import org.scalatest.{ Matchers, WordSpec, WordSpecLike }
+
 import scala.util.control.NoStackTrace
 import scala.concurrent.duration._
-
 import akka.actor.typed.SupervisorStrategy.Resume
+import akka.event.Logging
 
 object SupervisionSpec {
 
@@ -1210,6 +1211,30 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
       probe.expectMessage("pong")
     }
 
+    "log exceptions when logging is enabled and provided log level matches" in {
+      val probe = TestProbe[Event]("evt")
+      val behv = Behaviors
+        .supervise(targetBehavior(probe.ref))
+        .onFailure[Exc1](SupervisorStrategy.restart.withLoggingEnabled(true).withLogLevel(Logging.InfoLevel))
+      val ref = spawn(behv)
+      EventFilter.info(pattern = "exc-1", source = ref.path.toString, occurrences = 1).intercept {
+        ref ! Throw(new Exc1)
+        probe.expectMessage(ReceivedSignal(PreRestart))
+      }
+    }
+
+    "do not log exceptions when logging is enabled and provided log level does not match" in {
+      val probe = TestProbe[Event]("evt")
+      val behv = Behaviors
+        .supervise(targetBehavior(probe.ref))
+        .onFailure[Exc1](SupervisorStrategy.restart.withLoggingEnabled(true).withLogLevel(Logging.DebugLevel))
+      val ref = spawn(behv)
+      EventFilter.info(pattern = "exc-1", source = ref.path.toString, occurrences = 0).intercept {
+        ref ! Throw(new Exc1)
+        probe.expectMessage(ReceivedSignal(PreRestart))
+      }
+    }
+
   }
 
   val allStrategies = Seq(

akka-actor-typed/src/main/scala/akka/actor/typed/SupervisorStrategy.scala

@@ -5,9 +5,11 @@
 package akka.actor.typed
 
 import akka.annotation.InternalApi
+import akka.event.Logging
+import akka.event.Logging.LogLevel
+
 import scala.concurrent.duration.FiniteDuration
 import scala.concurrent.duration.Duration
-
 import akka.util.JavaDurationConverters._
 
 object SupervisorStrategy {
@@ -19,7 +21,7 @@ object SupervisorStrategy {
    * If the actor behavior is deferred and throws an exception on startup the actor is stopped
    * (restarting would be dangerous as it could lead to an infinite restart-loop)
    */
-  val resume: SupervisorStrategy = Resume(loggingEnabled = true)
+  val resume: SupervisorStrategy = Resume(loggingEnabled = true, logLevel = Logging.ErrorLevel)
 
   /**
    * Restart immediately without any limit on number of restart retries. A limit can be
@@ -34,7 +36,7 @@ object SupervisorStrategy {
   /**
    * Stop the actor
    */
-  val stop: SupervisorStrategy = Stop(loggingEnabled = true)
+  val stop: SupervisorStrategy = Stop(loggingEnabled = true, logLevel = Logging.ErrorLevel)
 
   /**
    * Scala API: It supports exponential back-off between the given `minBackoff` and
@@ -107,17 +109,21 @@ object SupervisorStrategy {
   /**
    * INTERNAL API
    */
-  @InternalApi private[akka] case class Resume(loggingEnabled: Boolean) extends SupervisorStrategy {
+  @InternalApi private[akka] case class Resume(loggingEnabled: Boolean, logLevel: LogLevel) extends SupervisorStrategy {
     override def withLoggingEnabled(enabled: Boolean): SupervisorStrategy =
       copy(loggingEnabled = enabled)
+    override def withLogLevel(level: LogLevel): SupervisorStrategy =
+      copy(logLevel = level)
   }
 
   /**
    * INTERNAL API
    */
-  @InternalApi private[akka] case class Stop(loggingEnabled: Boolean) extends SupervisorStrategy {
+  @InternalApi private[akka] case class Stop(loggingEnabled: Boolean, logLevel: LogLevel) extends SupervisorStrategy {
     override def withLoggingEnabled(enabled: Boolean) =
       copy(loggingEnabled = enabled)
+    override def withLogLevel(level: LogLevel): SupervisorStrategy =
+      copy(logLevel = level)
   }
 
   /**
@@ -139,6 +145,7 @@ object SupervisorStrategy {
       maxRestarts: Int,
       withinTimeRange: FiniteDuration,
       loggingEnabled: Boolean = true,
+      logLevel: LogLevel = Logging.ErrorLevel,
       stopChildren: Boolean = true,
       stashCapacity: Int = -1)
       extends RestartSupervisorStrategy
@@ -159,6 +166,8 @@ object SupervisorStrategy {
     override def withLoggingEnabled(enabled: Boolean): RestartSupervisorStrategy =
       copy(loggingEnabled = enabled)
 
+    override def withLogLevel(level: LogLevel): RestartSupervisorStrategy =
+      copy(logLevel = level)
   }
 
   /**
@@ -170,6 +179,7 @@ object SupervisorStrategy {
       randomFactor: Double,
       resetBackoffAfter: FiniteDuration,
       loggingEnabled: Boolean = true,
+      logLevel: LogLevel = Logging.ErrorLevel,
       maxRestarts: Int = -1,
       stopChildren: Boolean = true,
       stashCapacity: Int = -1)
@@ -195,13 +205,21 @@ object SupervisorStrategy {
 
     override def withLoggingEnabled(enabled: Boolean): BackoffSupervisorStrategy =
       copy(loggingEnabled = enabled)
+
+    override def withLogLevel(level: LogLevel): BackoffSupervisorStrategy =
+      copy(logLevel = logLevel)
+
   }
 }
 
 sealed abstract class SupervisorStrategy {
   def loggingEnabled: Boolean
+  def logLevel: LogLevel
 
   def withLoggingEnabled(enabled: Boolean): SupervisorStrategy
+
+  def withLogLevel(level: LogLevel): SupervisorStrategy
+
 }
 
 sealed abstract class RestartSupervisorStrategy extends SupervisorStrategy {
@@ -255,6 +273,8 @@ sealed abstract class RestartSupervisorStrategy extends SupervisorStrategy {
 
   override def withLoggingEnabled(enabled: Boolean): RestartSupervisorStrategy
 
+  override def withLogLevel(level: LogLevel): RestartSupervisorStrategy
+
 }
 
 sealed abstract class BackoffSupervisorStrategy extends SupervisorStrategy {
@@ -301,4 +321,6 @@ sealed abstract class BackoffSupervisorStrategy extends SupervisorStrategy {
 
   override def withLoggingEnabled(enabled: Boolean): BackoffSupervisorStrategy
 
+  override def withLogLevel(level: LogLevel): BackoffSupervisorStrategy
+
 }

akka-actor-typed/src/main/scala/akka/actor/typed/internal/Supervision.scala

@@ -77,7 +77,13 @@ private abstract class AbstractSupervisor[O, I, Thr <: Throwable](strategy: Supe
   def log(ctx: TypedActorContext[_], t: Throwable): Unit = {
     if (strategy.loggingEnabled) {
       val unwrapped = UnstashException.unwrap(t)
-      ctx.asScala.log.error(unwrapped, "Supervisor {} saw failure: {}", this, unwrapped.getMessage)
+      strategy.logLevel match {
+        case Logging.ErrorLevel =>
+          ctx.asScala.log.error(unwrapped, "Supervisor {} saw failure: {}", this, unwrapped.getMessage)
+        case Logging.WarningLevel =>
+          ctx.asScala.log.warning(unwrapped, "Supervisor {} saw failure: {}", this, unwrapped.getMessage)
+        case level => ctx.asScala.log.log(level, "Supervisor {} saw failure: {}", this, unwrapped.getMessage)
+      }
     }
   }