{"action":"created","issue":{"url":"https://api.github.com/repos/akka/reactive-kafka/issues/163","repository_url":"https://api.github.com/repos/akka/reactive-kafka","labels_url":"https://api.github.com/repos/akka/reactive-kafka/issues/163/labels{/name}","comments_url":"https://api.github.com/repos/akka/reactive-kafka/issues/163/comments","events_url":"https://api.github.com/repos/akka/reactive-kafka/issues/163/events","html_url":"https://github.com/akka/reactive-kafka/issues/163","id":162803997,"number":163,"title":"Producing messages during broker restart throws UnknownTopicOrPartitionException","user":{"login":"bfil","id":660513,"avatar_url":"https://avatars.githubusercontent.com/u/660513?v=3","gravatar_id":"","url":"https://api.github.com/users/bfil","html_url":"https://github.com/bfil","followers_url":"https://api.github.com/users/bfil/followers","following_url":"https://api.github.com/users/bfil/following{/other_user}","gists_url":"https://api.github.com/users/bfil/gists{/gist_id}","starred_url":"https://api.github.com/users/bfil/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bfil/subscriptions","organizations_url":"https://api.github.com/users/bfil/orgs","repos_url":"https://api.github.com/users/bfil/repos","events_url":"https://api.github.com/users/bfil/events{/privacy}","received_events_url":"https://api.github.com/users/bfil/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":0,"created_at":"2016-06-28T22:41:08Z","updated_at":"2016-06-29T05:35:15Z","closed_at":null,"body":"I have a very simple producer that produces a message every second using the new API (0.11-M3):\r\n\r\n```scala\r\nval decider: Supervision.Decider = {\r\n  case ex: Exception =>\r\n    log.error(ex, ex.getMessage)\r\n    Supervision.Resume\r\n}\r\n\r\nSource(1 to 100)\r\n  .throttle(1, 1 second, 1, ThrottleMode.shaping)\r\n  .map(_.toString)\r\n  .map(elem => {\r\n    log.info(s\"Producing: test-$elem\")\r\n    ProducerMessage.Message(new ProducerRecord[Array[Byte], String](topic, s\"test-$elem\"), s\"test-$elem\")\r\n  })\r\n  .via(Producer.flow(producerSettings).withAttributes(ActorAttributes.supervisionStrategy(decider)))\r\n  .map { result =>\r\n    val record = result.message.record\r\n    log.info(s\"Produced message: ${record.value}, topic ${record.topic}, partition: ${record.partition}, offset: ${result.offset}\")\r\n    result\r\n  }\r\n  .runWith(Sink.ignore)\r\n```\r\n\r\nMessages are dropped and not published during a broker restart.\r\nIt seems to be caused by temporarily outdated client metadata because it fails to be fetched.\r\n\r\nThis is a sample of the logs I'm getting by using the above producer (messages 5 to 10 are lost):\r\n\r\n```\r\n[INFO] [2016-06-28 23:53:45.338] [akka.actor.ActorSystemImpl] Producing: test-1\r\n[INFO] [2016-06-28 23:53:45.460] [akka.actor.ActorSystemImpl] Produced message: test-1, topic test, partition: null, offset: 29\r\n[INFO] [2016-06-28 23:53:46.356] [akka.actor.ActorSystemImpl] Producing: test-2\r\n[INFO] [2016-06-28 23:53:46.358] [akka.actor.ActorSystemImpl] Produced message: test-2, topic test, partition: null, offset: 30\r\n[INFO] [2016-06-28 23:53:47.356] [akka.actor.ActorSystemImpl] Producing: test-3\r\n[INFO] [2016-06-28 23:53:47.358] [akka.actor.ActorSystemImpl] Produced message: test-3, topic test, partition: null, offset: 31\r\n[INFO] [2016-06-28 23:53:48.356] [akka.actor.ActorSystemImpl] Producing: test-4\r\n[INFO] [2016-06-28 23:53:48.359] [akka.actor.ActorSystemImpl] Produced message: test-4, topic test, partition: null, offset: 32\r\n[INFO] [2016-06-28 23:53:49.355] [akka.actor.ActorSystemImpl] Producing: test-5\r\n[INFO] [2016-06-28 23:53:50.356] [akka.actor.ActorSystemImpl] Producing: test-6\r\n[INFO] [2016-06-28 23:53:51.355] [akka.actor.ActorSystemImpl] Producing: test-7\r\n[INFO] [2016-06-28 23:53:52.356] [akka.actor.ActorSystemImpl] Producing: test-8\r\n[INFO] [2016-06-28 23:53:53.356] [akka.actor.ActorSystemImpl] Producing: test-9\r\n[WARN] [2016-06-28 23:54:01.533] [o.apache.kafka.clients.NetworkClient] Error while fetching metadata with correlation id 6 : {test=LEADER_NOT_AVAILABLE}\r\n[ERROR] [2016-06-28 23:54:01.618] [akka.actor.ActorSystemImpl] This server does not host this topic-partition.\r\norg.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.\r\n[INFO] [2016-06-28 23:54:01.619] [akka.actor.ActorSystemImpl] Producing: test-10\r\n[ERROR] [2016-06-28 23:54:01.619] [akka.actor.ActorSystemImpl] This server does not host this topic-partition.\r\norg.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.\r\n[ERROR] [2016-06-28 23:54:01.619] [akka.actor.ActorSystemImpl] This server does not host this topic-partition.\r\norg.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.\r\n[ERROR] [2016-06-28 23:54:01.620] [akka.actor.ActorSystemImpl] This server does not host this topic-partition.\r\norg.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.\r\n[ERROR] [2016-06-28 23:54:01.620] [akka.actor.ActorSystemImpl] This server does not host this topic-partition.\r\norg.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.\r\n[ERROR] [2016-06-28 23:54:01.624] [akka.actor.ActorSystemImpl] This server does not host this topic-partition.\r\norg.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.\r\n[WARN] [2016-06-28 23:54:01.637] [o.apache.kafka.clients.NetworkClient] Error while fetching metadata with correlation id 9 : {test=LEADER_NOT_AVAILABLE}\r\n[INFO] [2016-06-28 23:54:02.357] [akka.actor.ActorSystemImpl] Producing: test-11\r\n[INFO] [2016-06-28 23:54:02.369] [akka.actor.ActorSystemImpl] Produced message: test-11, topic test, partition: null, offset: 33\r\n[INFO] [2016-06-28 23:54:03.357] [akka.actor.ActorSystemImpl] Producing: test-12\r\n[INFO] [2016-06-28 23:54:03.359] [akka.actor.ActorSystemImpl] Produced message: test-12, topic test, partition: null, offset: 34\r\n[INFO] [2016-06-28 23:54:04.356] [akka.actor.ActorSystemImpl] Producing: test-13\r\n[INFO] [2016-06-28 23:54:04.358] [akka.actor.ActorSystemImpl] Produced message: test-13, topic test, partition: null, offset: 35\r\n[INFO] [2016-06-28 23:54:05.358] [akka.actor.ActorSystemImpl] Producing: test-14\r\n[INFO] [2016-06-28 23:54:05.360] [akka.actor.ActorSystemImpl] Produced message: test-14, topic test, partition: null, offset: 36\r\n[INFO] [2016-06-28 23:54:06.356] [akka.actor.ActorSystemImpl] Producing: test-15\r\n[INFO] [2016-06-28 23:54:06.358] [akka.actor.ActorSystemImpl] Produced message: test-15, topic test, partition: null, offset: 37\r\n```\r\n\r\nIs this the expected behavior? Or should the fetch metadata failures be handled more gracefully by reactive-kafka?"},"comment":{"url":"https://api.github.com/repos/akka/reactive-kafka/issues/comments/229260929","html_url":"https://github.com/akka/reactive-kafka/issues/163#issuecomment-229260929","issue_url":"https://api.github.com/repos/akka/reactive-kafka/issues/163","id":229260929,"user":{"login":"iravid","id":283332,"avatar_url":"https://avatars.githubusercontent.com/u/283332?v=3","gravatar_id":"","url":"https://api.github.com/users/iravid","html_url":"https://github.com/iravid","followers_url":"https://api.github.com/users/iravid/followers","following_url":"https://api.github.com/users/iravid/following{/other_user}","gists_url":"https://api.github.com/users/iravid/gists{/gist_id}","starred_url":"https://api.github.com/users/iravid/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/iravid/subscriptions","organizations_url":"https://api.github.com/users/iravid/orgs","repos_url":"https://api.github.com/users/iravid/repos","events_url":"https://api.github.com/users/iravid/events{/privacy}","received_events_url":"https://api.github.com/users/iravid/received_events","type":"User","site_admin":false},"created_at":"2016-06-29T05:35:15Z","updated_at":"2016-06-29T05:35:15Z","body":"If you need a workaround, you can get the underlying client to handle this with retries by setting the `retries` property on the producer settings.\r\n\r\nNote that this can cause reordering of messages, so you'll also need to set `max.in.flight.requests.per.connection` to 1 to avoid that."}}
{"actor":{"display_login":"iravid"}}