Akka HTTP / Error Response entity was not subscribed after 1 second

Viewed 237

I searched the other StackOverflow question/answers towards this error, but couldn't find a hint for solving this problem.

The Akka HTTP application runs for like 5 hours under high workload without problems, and than I start to get multiple:

Response entity was not subscribed after 1 second. Make sure to read the response `entity` body or call `entity.discardBytes()` on it -- in case you deal with `HttpResponse`, use the shortcut `response.discardEntityBytes()`. GET /api/name123 Empty -> 200 OK Default(142 bytes)

and later

The connection actor has terminated. Stopping now.

The actor is only sending out API requests and afterwards forwards those responses to another actor if successfully, in case of failure, that request is added back to the todo stack and retried later. This is the main code:

private def makeApiRequest(id: String): Unit = {
    val url = UrlBuilder(id)
    val request = HttpRequest(method = HttpMethods.GET, uri = url)

    val f: Future[(StatusCode, String)] = Http(context.system)
      .singleRequest(request) 
      .flatMap(_.toStrict(2.seconds)) 
      .flatMap { resp =>
        Unmarshal(resp.entity).to[String].map((resp.status, _))
      }

    context.pipeToSelf(f) {
      case Success(response) =>
        API_HandleResponseSuccess(id, response._1, response._2) 

      case Failure(e) =>
        API_HandleResponseFailure(id, e.getMessage)  
    }
  }

I don't really understand why I get the "Response entity was not subscribed..." error, as I do Unmarshal(resp.entity).to[String] and thereby would think, that no .DiscardEntityBytes() is needed, or does it needs to be still included somehow?


Side information: Also confusing to me, why the CPU performance doesn't stay constant. enter image description here


Within the actor do I track the response times of each request and calculate the amount of max. parallel requests possible to handle with the given hardware conditions (restricted to a max max of 120 though) on a regular basis to account for API response time fluctuations, so there should be always enough room to make the requests without starving for that actor. In addition would that be the respective application.conf:

dispatcher-worker-io {
  type = Dispatcher
  executor = "thread-pool-executor"
  thread-pool-executor {
    fixed-pool-size = 120 
    keep-alive-time = 60s
    allow-core-timeout = off
  }
  shutdown-timeout = 60s 
  throughput = 1
}

...
akka.http.client.host-connection-pool.max-connections = 180
akka.http.client.host-connection-pool.max-open-requests = 256
akka.http.client.host-connection-pool.max-retries = 0

Any ideas on why I after 5 hours without problems start to get those exceptions mentioned above?

or

Has an idea of which part of above shared code might leads to this non-linear CPU performance?


I also made multiple of those long lasting hour runs, and it always ends out like this, somehow it's starving after 5 to 6 hours.


val AkkaVersion = "2.6.15"
val AkkaHttpVersion = "10.2.6"
1 Answers

Directly from the docs (https://doc.akka.io/docs/akka-http/current/client-side/request-level.html):

Always make sure you consume the response entity streams (of type Source[ByteString,Unit]). Connect the response entity Source to a Sink, or call response.discardEntityBytes() if you don’t care about the response entity.

Read the Implications of the streaming nature of Request/Response Entities section for more details.

If the application doesn’t subscribe to the response entity within akka.http.host-connection-pool.response-entity-subscription-timeout, the stream will fail with a TimeoutException: Response entity was not subscribed after ....

You need to .discardEntityBytes() in case of failure. Right now you only consume it on success.

Perhaps high CPU load is caused by all these unfreed resources on the JVM + retries of all the failures.

Related