I recently started creating a small web processing service using akka streams. It's pretty simple, I pull the URLs from redis, then load these URLs (they are images), later I process the images and click them on s3 and some json on redis.
I load many different images from several sites, I get a whole bunch of errors, such as 404, Unexpected shutdown, Response Content-Length 17951202 exceeds configured limit 8388608, EntityStreamException: Truncating and redirecting entity stream, When redirecting, I call requestWithRedirects with an address based on response location header.
The part responsible for downloading is something like this:
override lazy val http: HttpExt = Http() def requestWithRedirects(request: HttpRequest, retries: Int = 10)(implicit akkaSystem: ActorSystem, materializer: FlowMaterializer): Future[HttpResponse] = { TimeoutFuture(timeout, msg = "Download timed out!") { http.singleRequest(request) }.flatMap { response => handleResponse(request, response, retries) }.recoverWith { case e: Exception if retries > 0 => requestWithRedirects(request, retries = retries - 1) } }
TimeoutFuture is quite simple; it takes time and time. If the future takes longer than the timeout, it returns a different future with the exception of the timeout. I have a problem: after a while I get an error message:
Message: RuntimeException: Exceeded configured max-open-requests value of [128] akka.http.impl.engine.client.PoolInterfaceActor$$anonfun$receive$1.applyOrElse in PoolInterfaceActor.scala::109 akka.actor.Actor$class.aroundReceive in Actor.scala::467 akka.http.impl.engine.client.PoolInterfaceActor.akka$stream$actor$ActorSubscriber$$super$aroundReceive in PoolInterfaceActor.scala::46 akka.stream.actor.ActorSubscriber$class.aroundReceive in ActorSubscriber.scala::208 akka.http.impl.engine.client.PoolInterfaceActor.akka$stream$actor$ActorPublisher$$super$aroundReceive in PoolInterfaceActor.scala::46 akka.stream.actor.ActorPublisher$class.aroundReceive in ActorPublisher.scala::317 akka.http.impl.engine.client.PoolInterfaceActor.aroundReceive in PoolInterfaceActor.scala::46 akka.actor.ActorCell.receiveMessage in ActorCell.scala::516 akka.actor.ActorCell.invoke in ActorCell.scala::487 akka.dispatch.Mailbox.processMailbox in Mailbox.scala::238 akka.dispatch.Mailbox.run in Mailbox.scala::220 akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec in AbstractDispatcher.scala::397 scala.concurrent.forkjoin.ForkJoinTask.doExec in ForkJoinTask.java::260 scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask in ForkJoinPool.java::1339 scala.concurrent.forkjoin.ForkJoinPool.runWorker in ForkJoinPool.java::1979 scala.concurrent.forkjoin.ForkJoinWorkerThread.run in ForkJoinWorkerThread.java::107
I'm not sure what might be the problem, but I think I have some downloads that have not finished correctly, and they remain in some global connection pool for some time, causing the mentioned error. Any ideas what might cause the problem? Or how to try to find the root of the problem: I already tested 404 responses, and Content Content-Length exceeds ... errors, and they do not seem to be my intruders.
EDIT: Most likely the problem is with my TimeoutFuture. I fill it with an error, as described here https://stackoverflow.com/questions/984359/..but , in my opinion, the future that actually loads the image never ends, and it takes the resources of the connection pool.
I wonder why these settings have no effect in my case:
akka.http.client.connecting-timeout = 1 s akka.http.client.idle-timeout = 1 s akka.http.host-connection-pool.idle-timeout = 1 s
EDIT2:
Apparently timeouts are not yet supported. Here is my error report https://github.com/akka/akka/issues/17732#issuecomment-112315953