{"action":"closed","issue":{"url":"https://api.github.com/repos/akka/akka/issues/20902","repository_url":"https://api.github.com/repos/akka/akka","labels_url":"https://api.github.com/repos/akka/akka/issues/20902/labels{/name}","comments_url":"https://api.github.com/repos/akka/akka/issues/20902/comments","events_url":"https://api.github.com/repos/akka/akka/issues/20902/events","html_url":"https://github.com/akka/akka/issues/20902","id":164187667,"number":20902,"title":"superpool leaks connections on graph failure","user":{"login":"stejskal","id":991981,"avatar_url":"https://avatars.githubusercontent.com/u/991981?v=3","gravatar_id":"","url":"https://api.github.com/users/stejskal","html_url":"https://github.com/stejskal","followers_url":"https://api.github.com/users/stejskal/followers","following_url":"https://api.github.com/users/stejskal/following{/other_user}","gists_url":"https://api.github.com/users/stejskal/gists{/gist_id}","starred_url":"https://api.github.com/users/stejskal/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/stejskal/subscriptions","organizations_url":"https://api.github.com/users/stejskal/orgs","repos_url":"https://api.github.com/users/stejskal/repos","events_url":"https://api.github.com/users/stejskal/events{/privacy}","received_events_url":"https://api.github.com/users/stejskal/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/akka/akka/labels/bug","name":"bug","color":"e11d21"},{"url":"https://api.github.com/repos/akka/akka/labels/t:http","name":"t:http","color":"CCCCCC"}],"state":"closed","locked":false,"assignee":null,"assignees":[],"milestone":{"url":"https://api.github.com/repos/akka/akka/milestones/96","html_url":"https://github.com/akka/akka/milestones/2.4.11","labels_url":"https://api.github.com/repos/akka/akka/milestones/96/labels","id":1988052,"number":96,"title":"2.4.11","description":"","creator":{"login":"2m","id":422086,"avatar_url":"https://avatars.githubusercontent.com/u/422086?v=3","gravatar_id":"","url":"https://api.github.com/users/2m","html_url":"https://github.com/2m","followers_url":"https://api.github.com/users/2m/followers","following_url":"https://api.github.com/users/2m/following{/other_user}","gists_url":"https://api.github.com/users/2m/gists{/gist_id}","starred_url":"https://api.github.com/users/2m/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/2m/subscriptions","organizations_url":"https://api.github.com/users/2m/orgs","repos_url":"https://api.github.com/users/2m/repos","events_url":"https://api.github.com/users/2m/events{/privacy}","received_events_url":"https://api.github.com/users/2m/received_events","type":"User","site_admin":false},"open_issues":42,"closed_issues":17,"state":"open","created_at":"2016-09-07T06:29:32Z","updated_at":"2016-09-08T18:58:04Z","due_on":null,"closed_at":null},"comments":7,"created_at":"2016-07-06T22:17:13Z","updated_at":"2016-09-08T18:58:04Z","closed_at":"2016-09-08T18:58:04Z","body":"opened based on discussion in forum:\r\nhttps://groups.google.com/forum/#!topic/akka-user/MJJADiJZOrY\r\n\r\nScala version: 2.11.8\r\nakka version: 2.4.6\r\n\r\nI have a graph that is leaking http connections from an Akka-Http Superpool when the graph fails.  The general shape of the graph is:\r\n\r\nA ~> B ~> C ~> D ~> E ~> F\r\n\r\nThe stage actions are:  \r\nA emits HttpRequests   \r\nB is the superpool (created by Http().superPool())  \r\nC does some validation on the HttpResponse but leaves the ResponseEntity alone  \r\nD uses flatMapConcat to emit the ResponseEntity ByteString Sources  \r\nE Collects the ResponseEntity ByteStrings into a single chunk  \r\nF further processes the ByteString\r\n\r\nWhich all works fine if the graph completes successfully.  However if stage F fails and fails the graph then one connection is never released and eventually a materialization of the graph stalls forever waiting for stage B to process a request.\r\n\r\nWhat I believe is happening is that when stage F fails, the ResponseEntity Source in stage C has not been sinked and so the connection it occupies is not released.\r\n\r\nIs there a way to sink that ResponseEntity source on graph failure or alternatively, is there a better way to push a series of HttpRequests through a superPool that is aware of the entityBody and sinks on failure?\r\n\r\nHere is a reproducible example of this problem.\r\n\r\n```scala\r\nimport java.util.concurrent.Executors\r\n\r\nimport akka.NotUsed\r\nimport akka.actor.ActorSystem\r\nimport akka.http.scaladsl.Http\r\nimport akka.http.scaladsl.model._\r\nimport akka.stream.ActorMaterializer\r\nimport akka.stream.scaladsl.{Flow, Keep, Sink, Source}\r\nimport akka.util.ByteString\r\n\r\nimport scala.collection.immutable.Seq\r\nimport scala.concurrent.duration._\r\nimport scala.concurrent.{ExecutionContext, Future}\r\nimport scala.util.{Failure, Success, Try}\r\n\r\nobject Example\r\n{\r\n  implicit val system = ActorSystem(\"example\")\r\n  implicit val ec = ExecutionContext.fromExecutorService(Executors.newSingleThreadExecutor())\r\n  implicit val mat = ActorMaterializer()\r\n\r\n  var requestCount = 0\r\n  val pool = Http().superPool[NotUsed]()\r\n  val requestSource = Source.repeat((HttpRequest(uri = \"https://www.google.com/images/branding/googlelogo/2x/googlelogo_color_272x92dp.png\"), NotUsed))\r\n  val responseValidationFlow = Flow[(Try[HttpResponse], NotUsed)].collect\r\n  {\r\n    case (Success(response), _) =>\r\n      if (response.status == StatusCodes.OK) Success(response.entity)\r\n      else\r\n      {\r\n        response.entity.dataBytes.runWith(Sink.ignore)\r\n        Failure(new Exception(\"non 200 status\"))\r\n      }\r\n  }.collect\r\n  {\r\n    case Success(entity) => entity\r\n  }\r\n\r\n  val extractBodyFlowWithStrict = Flow[ResponseEntity].mapAsync(1)(entity => entity.toStrict(2 seconds).map(_.data))\r\n\r\n  val failableFlow = Flow[ByteString].map\r\n  {\r\n    imageBytes =>\r\n      requestCount += 1\r\n      if (requestCount % 2 == 0) throw new Exception(\"i blew up\")\r\n      println(imageBytes)\r\n      imageBytes\r\n  }\r\n\r\n  val sink: Sink[ByteString, Future[Seq[ByteString]]] = Sink.seq[ByteString]\r\n\r\n  val runnableGraph = requestSource\r\n    .via(pool)\r\n    .via(responseValidationFlow)\r\n    .via(extractBodyFlowWithStrict)\r\n    .via(failableFlow)\r\n    .toMat(sink)(Keep.right)\r\n\r\n  def main(args: Array[String]): Unit =\r\n  {\r\n    materialize\r\n  }\r\n\r\n  def materialize: Unit =\r\n  {\r\n    println(\"starting graph\")\r\n    runnableGraph.run().recover\r\n    {\r\n      case t =>\r\n        println(s\"graph failed ${t.getMessage}\")\r\n        materialize\r\n    }\r\n  }\r\n}\r\n```\r\n\r\nthanks,\r\nSpencer"}}
{"actor":{"display_login":"johanandren"}}