I am using Delayed restarts with backoff stage feature of Akka Streams which doesn't seem to work for me.
The code for my test is:
object Test {
import akka.stream.scaladsl.{ Flow, RestartFlow, Sink, Source }
import scala.concurrent.duration._
import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
implicit val actorSystem = ActorSystem()
implicit val mat = ActorMaterializer()
def main(args: Array[String]): Unit = {
val source = Source(1 to 10)
val flow = Flow[Int].map { x =>
println(s"Processing: $x")
if (x != 6) {
x * 2
} else throw new RuntimeException("Baam!!")
}
val restartFlow = RestartFlow.onFailuresWithBackoff(10.milliseconds, 20.milliseconds, 0.2, 3)(() => flow)
source.via(restartFlow).to(Sink.ignore).run()
}
}
I expect the exception to happen three times as the max restarts is equal to 3. But, on the contrary I see exception only once. What am I missing here?
Output of running it:
Processing: 1
Processing: 2
Processing: 3
Processing: 4
Processing: 5
Processing: 6
[ERROR] [03/20/2018 16:13:59.167] [default-akka.actor.default-dispatcher-2] [RestartWithBackoffFlow(akka://default)] Restarting graph due to failure
java.lang.RuntimeException: Baam!!
at Test$.$anonfun$main$1(Test.scala:18)
at scala.runtime.java8.JFunction1$mcII$sp.apply(JFunction1$mcII$sp.java:12)
at akka.stream.impl.fusing.Map$$anon$9.onPush(Ops.scala:53)
at akka.stream.impl.fusing.GraphInterpreter.processPush(GraphInterpreter.scala:519)
at akka.stream.impl.fusing.GraphInterpreter.processEvent(GraphInterpreter.scala:482)
at akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:378)
at akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:585)
at akka.stream.impl.fusing.GraphInterpreterShell$AsyncInput.execute(ActorGraphInterpreter.scala:469)
at akka.stream.impl.fusing.GraphInterpreterShell.processEvent(ActorGraphInterpreter.scala:560)
at akka.stream.impl.fusing.ActorGraphInterpreter.akka$stream$impl$fusing$ActorGraphInterpreter$$processEvent(ActorGraphInterpreter.scala:742)
at akka.stream.impl.fusing.ActorGraphInterpreter.akka$stream$impl$fusing$ActorGraphInterpreter$$shortCircuitBatch(ActorGraphInterpreter.scala:732)
at akka.stream.impl.fusing.ActorGraphInterpreter.preStart(ActorGraphInterpreter.scala:726)
at akka.actor.Actor.aroundPreStart(Actor.scala:528)
at akka.actor.Actor.aroundPreStart$(Actor.scala:528)
at akka.stream.impl.fusing.ActorGraphInterpreter.aroundPreStart(ActorGraphInterpreter.scala:667)
at akka.actor.ActorCell.create(ActorCell.scala:654)
at akka.actor.ActorCell.invokeAll$1(ActorCell.scala:525)
at akka.actor.ActorCell.systemInvoke(ActorCell.scala:547)
at akka.dispatch.Mailbox.processAllSystemMessages(Mailbox.scala:282)
at akka.dispatch.Mailbox.run(Mailbox.scala:223)
at akka.dispatch.Mailbox.exec(Mailbox.scala:234)
at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
