Exceeded configured max-open-requests

Question

recently I started to build some small web processing service using akka streams. It's quite simple, I'm pulling urls from redis, then I'm downloading those urls(they are images) later I'm processing images, and pushing them to s3 and some json to redis.

I'm downloading lot of different kinds of images from multiple sites, I'm getting whole bunch of errors like 404, Unexpected disconnect , Response Content-Length 17951202 exceeds the configured limit of 8388608, EntityStreamException: Entity stream truncation and redirects. With redirects I'm invoking requestWithRedirects with address founded in location header of response.

Part responsible for downloading is pretty much 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 future and timeout. If future takes longer than timeout it returns other future with timeout exception. The problem I'm having is: after some time I'm getting an error:

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 could be the problem but I think I have some downloads that were not finished properly and they stay in some global pool of connections after a while causing mentioned error. Any ideas what could be causing the problem? Or how to try find root of the problem: I already tested 404 responses, and Response Content-Length exceeds... errors, and they doesn't seem to be my troublemakers.

EDIT: Most likely the problem is with my TimeoutFuture. I'm filling it with error as described here https://stackoverflow.com/a/29330010/2963977 but in my opinion future that is actually downloading an image never completes and it's taking my connection pool resources.

I wonder why those settings doesn't have any impact 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 supported yet. Here is my bug report https://github.com/akka/akka/issues/17732#issuecomment-112315953