akka/akka-core
View on GitHubAbruptTerminationException is silenced when using RestartSource.withBackoff()
Open
#27,193 opened on Jun 20, 2019
1 - triagedhelp wanted
Repository metrics
- Stars
- (13,277 stars)
- PR merge metrics
- (Avg merge 8d 19h) (10 merged PRs in 30d)
Description
Description
There is a stream wrapped with RestartSource.withBackoff().
-
When ordinary failures occur within the stream there are nice warnings logged:
Restarting graph due to failure. stack_trace: …
And the stream is restarted.
-
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()