{"action":"created","issue":{"url":"https://api.github.com/repos/akka/akka-http/issues/416","repository_url":"https://api.github.com/repos/akka/akka-http","labels_url":"https://api.github.com/repos/akka/akka-http/issues/416/labels{/name}","comments_url":"https://api.github.com/repos/akka/akka-http/issues/416/comments","events_url":"https://api.github.com/repos/akka/akka-http/issues/416/events","html_url":"https://github.com/akka/akka-http/issues/416","id":184300403,"number":416,"title":"Response entity failures kill client pool","user":{"login":"jrudolph","id":9868,"avatar_url":"https://avatars.githubusercontent.com/u/9868?v=3","gravatar_id":"","url":"https://api.github.com/users/jrudolph","html_url":"https://github.com/jrudolph","followers_url":"https://api.github.com/users/jrudolph/followers","following_url":"https://api.github.com/users/jrudolph/following{/other_user}","gists_url":"https://api.github.com/users/jrudolph/gists{/gist_id}","starred_url":"https://api.github.com/users/jrudolph/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jrudolph/subscriptions","organizations_url":"https://api.github.com/users/jrudolph/orgs","repos_url":"https://api.github.com/users/jrudolph/repos","events_url":"https://api.github.com/users/jrudolph/events{/privacy}","received_events_url":"https://api.github.com/users/jrudolph/received_events","type":"User","site_admin":false},"labels":[{"id":430145635,"url":"https://api.github.com/repos/akka/akka-http/labels/1%20-%20triaged","name":"1 - triaged","color":"5319e7","default":false},{"id":427494657,"url":"https://api.github.com/repos/akka/akka-http/labels/bug","name":"bug","color":"ee0701","default":true},{"id":438995535,"url":"https://api.github.com/repos/akka/akka-http/labels/t:http:client","name":"t:http:client","color":"ededed","default":false},{"id":439004097,"url":"https://api.github.com/repos/akka/akka-http/labels/t:http:core","name":"t:http:core","color":"ededed","default":false}],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":{"url":"https://api.github.com/repos/akka/akka-http/milestones/14","html_url":"https://github.com/akka/akka-http/milestones/FIRST_STABLE","labels_url":"https://api.github.com/repos/akka/akka-http/milestones/14/labels","id":2074877,"number":14,"title":"FIRST_STABLE","description":"","creator":{"login":"ktoso","id":120979,"avatar_url":"https://avatars.githubusercontent.com/u/120979?v=3","gravatar_id":"","url":"https://api.github.com/users/ktoso","html_url":"https://github.com/ktoso","followers_url":"https://api.github.com/users/ktoso/followers","following_url":"https://api.github.com/users/ktoso/following{/other_user}","gists_url":"https://api.github.com/users/ktoso/gists{/gist_id}","starred_url":"https://api.github.com/users/ktoso/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ktoso/subscriptions","organizations_url":"https://api.github.com/users/ktoso/orgs","repos_url":"https://api.github.com/users/ktoso/repos","events_url":"https://api.github.com/users/ktoso/events{/privacy}","received_events_url":"https://api.github.com/users/ktoso/received_events","type":"User","site_admin":false},"open_issues":10,"closed_issues":8,"state":"open","created_at":"2016-10-17T16:00:36Z","updated_at":"2016-10-26T16:20:23Z","due_on":null,"closed_at":null},"comments":7,"created_at":"2016-10-20T18:06:29Z","updated_at":"2016-10-30T12:22:16Z","closed_at":null,"body":"It was really hard to find out how problems with a response are fed back into the pool so that at the end the complete pool is shutdown.\r\n\r\nI changed GraphStageLogic.failStage to wrap all exceptions with more information to get at the trace of how the error propagated like this:\r\n\r\n```scala\r\n  final def failStage(ex: Throwable): Unit = {\r\n    def stageName: String = try {\r\n      val outerField = getClass.getDeclaredField(\"$outer\")\r\n      outerField.setAccessible(true)\r\n      outerField.get(this) match {\r\n        case MapAsyncUnordered(_, func) ⇒ s\"MapAsyncUnordered(${func.getClass})\"\r\n        case Map(func)                  ⇒ s\"Map(${func.getClass})\"\r\n        case Collect(func)              ⇒ s\"Collect(${func.getClass})\"\r\n        case _                          ⇒ getClass.toString\r\n      }\r\n    } catch {\r\n      case x ⇒ getClass.toString\r\n    }\r\n\r\n    val newEx = new RuntimeException(s\"Failing stage in $stageName\", ex).fillInStackTrace()\r\n    // use newEx instead of ex\r\n  }\r\n```\r\n\r\nYou then get a stack trace like this (stripped the actual code lines) if while receiving a response entity the connection breaks:\r\n\r\n```\r\njava.lang.RuntimeException: Failing stage in class akka.http.impl.engine.client.PoolSlot$SlotProcessor$$anon$3\r\nCaused by: java.lang.RuntimeException: Failing stage in class akka.stream.impl.fusing.Log$$anon$23\r\nCaused by: java.lang.RuntimeException: Failing stage in class akka.http.impl.engine.client.PoolConductor$Route$$anon$4\r\nCaused by: java.lang.RuntimeException: Failing stage in class akka.http.impl.engine.client.PoolConductor$SlotSelector$$anon$1\r\nCaused by: java.lang.RuntimeException: Failing stage in MapAsyncUnordered(class akka.http.impl.engine.client.PoolConductor$$anonfun$apply$2$$anonfun$4)\r\nCaused by: java.lang.RuntimeException: Failing stage in class akka.http.scaladsl.model.HttpEntity$Limitable$$anon$1\r\nCaused by: java.lang.RuntimeException: Failing stage in Collect(class akka.http.impl.engine.parsing.HttpMessageParser$$anonfun$defaultEntity$1$$anonfun$1)\r\nCaused by: java.lang.RuntimeException: Failing stage in class akka.stream.impl.fusing.SubSource$$anon$4\r\nCaused by: java.lang.RuntimeException: Failing stage in class akka.stream.impl.fusing.StatefulMapConcat$$anon$25\r\nCaused by: java.lang.RuntimeException: Failing stage in class akka.http.impl.engine.client.OutgoingConnectionBlueprint$ResponseParsingMerge$$anon$6\r\nCaused by: java.lang.RuntimeException: Failing stage in Collect(class akka.http.impl.engine.client.OutgoingConnectionBlueprint$$anonfun$6$$anonfun$5)\r\nCaused by: java.lang.RuntimeException: Failing stage in class akka.stream.impl.Timers$IdleTimeoutBidi$$anon$7\r\n.RuntimeException: Failing stage in class akka.stream.impl.fusing.GraphStages$Detacher$$anon$2\r\nCaused by: java.lang.RuntimeException: Failing stage in class \r\nCaused by: akka.stream.StreamTcpException: The connection closed with error: Connection reset by peer\r\n```\r\n\r\nThe interesting connection is\r\n\r\n```\r\nCaused by: java.lang.RuntimeException: Failing stage in MapAsyncUnordered(class akka.http.impl.engine.client.PoolConductor$$anonfun$apply$2$$anonfun$4)\r\nCaused by: java.lang.RuntimeException: Failing stage in class akka.http.scaladsl.model.HttpEntity$Limitable$$anon$1\r\n```\r\n\r\nwhere an entity error suddenly gets reported as an error in the RawSlotEvent channel. The `mapAsyncUnordered` is the one from this line: https://github.com/akka/akka-http/blob/master/akka-http-core/src/main/scala/akka/http/impl/engine/client/PoolConductor.scala#L78-L78\r\n\r\nTurns out that PoolSlot does\r\n\r\n```scala\r\nval (entity, whenCompleted) = HttpEntity.captureTermination(response.entity)\r\n          //...\r\n          push(eventsOut, SlotEvent.RequestCompletedFuture(whenCompleted.map(_ ⇒ SlotEvent.RequestCompleted(slotIx))))\r\n```\r\n\r\nto capture termination of a response entity to find out that a connection is free again in the Future `whenCompleted`.\r\n\r\nThis is then handled in the linked-above `mapAsyncUnordered`:\r\n\r\n```scala\r\n      val flatten = Flow[RawSlotEvent].mapAsyncUnordered(slotSettings.maxSlots) {\r\n        case x: SlotEvent.Disconnected                ⇒ FastFuture.successful(x)\r\n        case SlotEvent.RequestCompletedFuture(future) ⇒ future\r\n        case x: SlotEvent.ConnectedEagerly            ⇒ FastFuture.successful(x)\r\n        case x                                        ⇒ throw new IllegalStateException(\"Unexpected \" + x)\r\n      }\r\n```\r\n\r\nSo, the `mapAsyncUnordered` will escalate an error in the entity stream to an error on the pool level which will subsequently kill the complete pool."},"comment":{"url":"https://api.github.com/repos/akka/akka-http/issues/comments/257147991","html_url":"https://github.com/akka/akka-http/issues/416#issuecomment-257147991","issue_url":"https://api.github.com/repos/akka/akka-http/issues/416","id":257147991,"user":{"login":"gosubpl","id":7828991,"avatar_url":"https://avatars.githubusercontent.com/u/7828991?v=3","gravatar_id":"","url":"https://api.github.com/users/gosubpl","html_url":"https://github.com/gosubpl","followers_url":"https://api.github.com/users/gosubpl/followers","following_url":"https://api.github.com/users/gosubpl/following{/other_user}","gists_url":"https://api.github.com/users/gosubpl/gists{/gist_id}","starred_url":"https://api.github.com/users/gosubpl/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/gosubpl/subscriptions","organizations_url":"https://api.github.com/users/gosubpl/orgs","repos_url":"https://api.github.com/users/gosubpl/repos","events_url":"https://api.github.com/users/gosubpl/events{/privacy}","received_events_url":"https://api.github.com/users/gosubpl/received_events","type":"User","site_admin":false},"created_at":"2016-10-30T12:22:16Z","updated_at":"2016-10-30T12:22:16Z","body":"@jrudolph please let me know if I am going in the right direction. The obvious fix (which makes the tests pass) would be:\r\n```\r\n          val completedF : Future[PoolSlot.SlotEvent.RequestCompleted] =\r\n            whenCompleted.map(_ ⇒ SlotEvent.RequestCompleted(slotIx))\r\n              .recoverWith{case _ => FastFuture.successful(SlotEvent.RequestCompleted(slotIx))}\r\n          push(eventsOut, SlotEvent.RequestCompletedFuture(completedF))\r\n```\r\nSo this would mean that the request completed, no matter whether it completed successfully or not. For sure some logging should be added here to let _ipsos custodes_ see what has happened.\r\nWha do you think?"}}
{"actor":{"display_login":"gosubpl"}}