{"action":"created","issue":{"url":"https://api.github.com/repos/akka/reactive-kafka/issues/200","repository_url":"https://api.github.com/repos/akka/reactive-kafka","labels_url":"https://api.github.com/repos/akka/reactive-kafka/issues/200/labels{/name}","comments_url":"https://api.github.com/repos/akka/reactive-kafka/issues/200/comments","events_url":"https://api.github.com/repos/akka/reactive-kafka/issues/200/events","html_url":"https://github.com/akka/reactive-kafka/issues/200","id":171857789,"number":200,"title":"Death of Kafka coordinator is not propagated to the stream's termination watch","user":{"login":"slavaschmidt","id":322297,"avatar_url":"https://avatars.githubusercontent.com/u/322297?v=3","gravatar_id":"","url":"https://api.github.com/users/slavaschmidt","html_url":"https://github.com/slavaschmidt","followers_url":"https://api.github.com/users/slavaschmidt/followers","following_url":"https://api.github.com/users/slavaschmidt/following{/other_user}","gists_url":"https://api.github.com/users/slavaschmidt/gists{/gist_id}","starred_url":"https://api.github.com/users/slavaschmidt/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/slavaschmidt/subscriptions","organizations_url":"https://api.github.com/users/slavaschmidt/orgs","repos_url":"https://api.github.com/users/slavaschmidt/repos","events_url":"https://api.github.com/users/slavaschmidt/events{/privacy}","received_events_url":"https://api.github.com/users/slavaschmidt/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":1,"created_at":"2016-08-18T09:31:56Z","updated_at":"2016-08-18T09:37:24Z","closed_at":null,"body":"I'm not sure if it's related to the #150\r\n\r\nFor following consumer utilising at-most-once semantics:\r\n``` scala\r\n  lazy val source = Consumer.committableSource(settings, topics)\r\n    .mapAsyncUnordered(kafkaParallelism) { msg =>\r\n      val _ = msg.committableOffset.commitScaladsl()\r\n      Future.successful(msg)\r\n    }\r\n```\r\ni'm registering a termination watch like this:\r\n`flow.watchTermination()(onStreamTermination).runWith(source, sink)`\r\n\r\nNow, if I shut down Kafka during the long-running test, I can see it is properly detected by observing following in logs:\r\n`INFO  o.a.k.c.c.i.AbstractCoordinator - Marking the coordinator localhost:9092 (id: 2147483647 rack: null) dead for group archiver`\r\n\r\nBut the termination watch never get's called! \r\n\r\nI'd expect that death of the source is propagated to the stream. If it is wrong expectation, what is a correct way of handling this situation?\r\n"},"comment":{"url":"https://api.github.com/repos/akka/reactive-kafka/issues/comments/240673423","html_url":"https://github.com/akka/reactive-kafka/issues/200#issuecomment-240673423","issue_url":"https://api.github.com/repos/akka/reactive-kafka/issues/200","id":240673423,"user":{"login":"13h3r","id":435085,"avatar_url":"https://avatars.githubusercontent.com/u/435085?v=3","gravatar_id":"","url":"https://api.github.com/users/13h3r","html_url":"https://github.com/13h3r","followers_url":"https://api.github.com/users/13h3r/followers","following_url":"https://api.github.com/users/13h3r/following{/other_user}","gists_url":"https://api.github.com/users/13h3r/gists{/gist_id}","starred_url":"https://api.github.com/users/13h3r/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/13h3r/subscriptions","organizations_url":"https://api.github.com/users/13h3r/orgs","repos_url":"https://api.github.com/users/13h3r/repos","events_url":"https://api.github.com/users/13h3r/events{/privacy}","received_events_url":"https://api.github.com/users/13h3r/received_events","type":"User","site_admin":false},"created_at":"2016-08-18T09:37:24Z","updated_at":"2016-08-18T09:37:24Z","body":"try this:\r\n\r\n```scala\r\n  lazy val source = Consumer.committableSource(settings, topics)\r\n    .mapAsyncUnordered(kafkaParallelism) { msg =>\r\n      msg.committableOffset.commitScaladsl().map(_ => msg)\r\n    }\r\n```"}}
{"actor":{"display_login":"13h3r"}}