{"action":"closed","issue":{"url":"https://api.github.com/repos/gzoller/LateKafka/issues/5","repository_url":"https://api.github.com/repos/gzoller/LateKafka","labels_url":"https://api.github.com/repos/gzoller/LateKafka/issues/5/labels{/name}","comments_url":"https://api.github.com/repos/gzoller/LateKafka/issues/5/comments","events_url":"https://api.github.com/repos/gzoller/LateKafka/issues/5/events","html_url":"https://github.com/gzoller/LateKafka/issues/5","id":150801107,"number":5,"title":"Is there a way to using Producer.commitableSink after a flow?","user":{"login":"MadeInChina","id":1201118,"avatar_url":"https://avatars.githubusercontent.com/u/1201118?v=3","gravatar_id":"","url":"https://api.github.com/users/MadeInChina","html_url":"https://github.com/MadeInChina","followers_url":"https://api.github.com/users/MadeInChina/followers","following_url":"https://api.github.com/users/MadeInChina/following{/other_user}","gists_url":"https://api.github.com/users/MadeInChina/gists{/gist_id}","starred_url":"https://api.github.com/users/MadeInChina/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/MadeInChina/subscriptions","organizations_url":"https://api.github.com/users/MadeInChina/orgs","repos_url":"https://api.github.com/users/MadeInChina/repos","events_url":"https://api.github.com/users/MadeInChina/events{/privacy}","received_events_url":"https://api.github.com/users/MadeInChina/received_events","type":"User","site_admin":false},"labels":[],"state":"closed","locked":false,"assignee":null,"milestone":null,"comments":1,"created_at":"2016-04-25T09:28:15Z","updated_at":"2016-04-26T06:30:02Z","closed_at":"2016-04-26T06:30:02Z","body":"Hi \r\n just see your code from gitter, what's i want to do is that deal message from kafka after that repush to kafka. i want ask suggestion or how to achieve as the code below.\r\nThanks.\r\n```\r\n  val src = Consumer.atMostOnceSource(consumerSettings.withClientId(\"client\" + id))\r\n  val commit = Flow[In].mapAsync(1) { msg =>\r\n    Future {\r\n      // msg.committableOffset.commit()\r\n      syncInc()\r\n      if (count == num) {\r\n        println(s\"[$id] Time ($count): \" + (count / ((System.currentTimeMillis() - now) / 1000)) + \" TPS\")\r\n      }\r\n    }\r\n  }\r\n  val work = Flow[In].map { i => i } // a dummy step where real \"work\" would happen\r\n\r\n  // src ~> work ~> commit ~> Sink.ignore\r\n  src ~> work ~> commit ~> Producer.commitableSink(producerSettings) \r\n  ClosedShape\r\n})\r\n```"}}