Hi Akka Team, All, I have few custom stages, recently i find out that one of them under certain conditions was not properly completing graph stage, which was causing graph to run forever. I have timeouts attached to graph and would expect it to fail with timeout but it never happened. I created a simple reproducer (below).
I have a BrokenStage that is not completing stage in onUpstreamFinish method and i have a graph which has idleTimeout, completionTimeout and killSwitch attached. None of this has an effect on the graph it fails with future timeout. I understand that this is developer issue and if i move any of timeouts after broken stage it will work as expected. My questions would be: 1. I would still expect killSwitch or completion timeout to fail the graph, why it is not a case? 2. how i can detect such things / safely attach timeouts to graphs? Reproducer: import akka.actor.ActorSystem import akka.stream._ import akka.stream.scaladsl.{Keep, Sink, Source} import akka.stream.stage.{GraphStage, GraphStageLogic, InHandler, OutHandler} import scala.concurrent.Await import scala.concurrent.duration._ case class BrokenStage[A](name: String) extends GraphStage[FlowShape[A, A]] { val in = Inlet[A]("in") val out = Outlet[A]("out") override val shape: FlowShape[A, A] = FlowShape(in, out) override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) { setHandler(in, new InHandler { override def onPush(): Unit = { println(s"$name: onPush") push(out, grab(in)) } override def onUpstreamFinish(): Unit = { println(s"$name: onUpstreamFinish") } override def onUpstreamFailure(ex: Throwable): Unit = { println(s"$name: onUpstreamFailure: ${ex.getMessage}") super.onUpstreamFailure(ex) } }) setHandler(out, new OutHandler { override def onPull(): Unit = { println(s"$name: onPull") pull(in) } override def onDownstreamFinish(): Unit = { println(s"$name: onDownstreamFinish") super.onDownstreamFinish() } }) } } object Example { implicit val actorSystem = ActorSystem() implicit val ec = actorSystem.dispatcher implicit val materializer = ActorMaterializer() def main(args: Array[String]): Unit = { val source = Source.empty[Int] val sink = Sink.last[Int] val (killSwitch, last) = source .idleTimeout(10.second) .completionTimeout(20.seconds) .viaMat(KillSwitches.single)(Keep.right) .via(new BrokenStage("after")) .toMat(sink)(Keep.both) .run() Thread.sleep(1000) killSwitch.abort(new RuntimeException("boom!")) Await.result(last, 25.second) sys.exit(-1) } } -- >>>>>>>>>> Read the docs: http://akka.io/docs/ >>>>>>>>>> Check the FAQ: >>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user --- You received this message because you are subscribed to the Google Groups "Akka User List" group. To unsubscribe from this group and stop receiving emails from it, send an email to akka-user+unsubscr...@googlegroups.com. To post to this group, send email to akka-user@googlegroups.com. Visit this group at https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.