Akka RestartSource does not restart

user_13956589 0
Viewed 339
object TestSource {
  implicit val ec = ExecutionContext.global

  def main(args: Array[String]): Unit = {
    def buildSource = {
      println("fresh")
      Source(List(() => 1,() => 2,() => 3,() => {
        println("crash")
      throw new RuntimeException(":(((")
      }))
    }
    val restarting = RestartSource.onFailuresWithBackoff(
      minBackoff = Duration(1, SECONDS) ,
      maxBackoff = Duration(1, SECONDS),
      randomFactor = 0.0,
      maxRestarts = 10
    )(() => {
      buildSource
    })

     implicit val actorSystem: ActorSystem           = ActorSystem()
     implicit val executionContext: ExecutionContext = actorSystem.dispatcher

    restarting.runWith(Sink.foreach(e => println(e())))

  }
}

The code above prints: 1,2,3, crash Why does my source not restart? This is pretty much a 1:1 copy of the official documentation.

edit:

I also tried

    val rs = RestartSink.withBackoff[() => Int](
      Duration(1, SECONDS),
      Duration(1, SECONDS),
      0.0,
      10
    )(_)
    val rsDone = rs(() => {
      println("???")
      Sink.foreach(e => println(e()))
    })
    restarting.runWith(rsDone)

but still get no restarts

akka-stream akka scala
3 Answers

This is because the exception is triggered outside of the buildSource Source in the Sink.foreach when you call the functions emitted from the Source.

Try this:

    val restarting = RestartSource.onFailuresWithBackoff(
      minBackoff = Duration(1, SECONDS) ,
      maxBackoff = Duration(1, SECONDS),
      randomFactor = 0.0,
      maxRestarts = 10
      )(() => {
        buildSource
         .map(e => e()) //call the functions inside the RestartSource
      })

That way your exception will happen inside the inner Source wrapped by RestartSource and the restarting mechanism will kick in.

user_9441740

The source doesn't restart because your source never fails, therefore never needs to restart.

The exception gets thrown when Sink.foreach evaluates the function it received.

As artur noted, if you can move the failing bit into the source, you can wrap everything up to the sink in the RestartSource.

While it won't help for this contrived example (as restarting a sink doesn't result in resending previously sent messages), wrapping the sink in a RestartSink may be useful in real-world cases where this sort of thing can happen (off the top of my head, streams from Kafka blowing up because the offset commit in a sink failed (e.g. after a rebalance) should be an example of such a case).

An alternative, if you want to restart the whole stream if any part fails, and the stream materializes as a Future, you can implement retry-with-backoff on the failed future.

user_56412440

Source just never crashes, as already said here. You are actually crashing you sink, not a source with this statement e => e()

this happens when applying lambda above to last element of source:

java.lang.RuntimeException: :(((

Here's the same stream without unhandled exception in sink:

... RestartSource.withBackoff( ...

restarting.runWith(
  Sink.foreach(e => {
    def i: Int = try{ e() } catch {
      case t: Throwable =>
        println(t)
        -1
    }
    println(i)
  })
)

Works perfectly.

user_80354240
Related