AbruptTerminationException is silenced when using RestartSource.withBackoff()

Description

There is a stream wrapped with RestartSource.withBackoff().

1. When ordinary failures occur within the stream there are nice warnings logged:

   Restarting graph due to failure. stack_trace: …

   And the stream is restarted.

2. But when the stream is abruptly stopped with AbruptTerminationException – there is no any warnings logged.

Possible cause

No failure logging can be seen here: https://github.com/akka/akka/blob/master/akka-stream/src/main/scala/akka/stream/scaladsl/RestartFlow.scala#L325

Code sample

class Supervisor(stream: Source[_, _]) extends Actor {

  override def preStart(): Unit = {
    restartingOnCrash(stream)
      .runWith(Sink.ignore)
      .map(_ => Stopped)
      .pipeTo(self)
    }

  override def receive: Receive = {
    case Stopped           =>
      log.info("Stream {} stopped", name)
      context.stop(self)
    case Status.Failure(t) =>
      log.error(t, "Stream {} crashed unexpectedly", name)
      context.stop(self)
  }

  private def restartingOnCrash[Out](stream: Source[Out, _]): Source[Out, _] =
    RestartSource.withBackoff(
      minBackoff = 1.second,
      maxBackoff = 10.seconds,
      randomFactor = 0.2d,
    ) { () =>
      stream.watchTermination() {
        case (_, f) => f.failed foreach { failure =>
          log.error(failure, "Stream {} will be restarted due to the failure", name)
        }
      }
    }
  }
}

actorSystem.actorOf(Props(Supervisor(Source.repeated(1))))
actorSystem.shutdown()