{"action":"created","issue":{"url":"https://api.github.com/repos/jacklund/kafka_ex/issues/31","labels_url":"https://api.github.com/repos/jacklund/kafka_ex/issues/31/labels{/name}","comments_url":"https://api.github.com/repos/jacklund/kafka_ex/issues/31/comments","events_url":"https://api.github.com/repos/jacklund/kafka_ex/issues/31/events","html_url":"https://github.com/jacklund/kafka_ex/issues/31","id":96075449,"number":31,"title":"Stream consumer not starting ","user":{"login":"sulphur","id":186265,"avatar_url":"https://avatars.githubusercontent.com/u/186265?v=3","gravatar_id":"","url":"https://api.github.com/users/sulphur","html_url":"https://github.com/sulphur","followers_url":"https://api.github.com/users/sulphur/followers","following_url":"https://api.github.com/users/sulphur/following{/other_user}","gists_url":"https://api.github.com/users/sulphur/gists{/gist_id}","starred_url":"https://api.github.com/users/sulphur/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sulphur/subscriptions","organizations_url":"https://api.github.com/users/sulphur/orgs","repos_url":"https://api.github.com/users/sulphur/repos","events_url":"https://api.github.com/users/sulphur/events{/privacy}","received_events_url":"https://api.github.com/users/sulphur/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":5,"created_at":"2015-07-20T14:23:31Z","updated_at":"2015-07-30T08:50:42Z","closed_at":null,"body":"Do you guys have some guidelines for creating the consumer. I spotted that when i lunch a consumer directly on app start it is unable to find the topic. From what i succeed to see it might be linked to the fact that it can't find a broker. In ```KafkaEx.Protocol.ConsumerMetadata``` on line 7 the ```consumer_group_metadata.coordinator_host``` is nil\r\n\r\nIf i create a GenServer after few seconds and launch the consumer after some time i suceed to consume the messages. So should i delalay the start of the consumer or maybe it is linked to something i didn't do ? ( I basically followed the readme to create a stream consumer)\r\n\r\n "},"comment":{"url":"https://api.github.com/repos/jacklund/kafka_ex/issues/comments/126229994","html_url":"https://github.com/jacklund/kafka_ex/issues/31#issuecomment-126229994","issue_url":"https://api.github.com/repos/jacklund/kafka_ex/issues/31","id":126229994,"user":{"login":"sulphur","id":186265,"avatar_url":"https://avatars.githubusercontent.com/u/186265?v=3","gravatar_id":"","url":"https://api.github.com/users/sulphur","html_url":"https://github.com/sulphur","followers_url":"https://api.github.com/users/sulphur/followers","following_url":"https://api.github.com/users/sulphur/following{/other_user}","gists_url":"https://api.github.com/users/sulphur/gists{/gist_id}","starred_url":"https://api.github.com/users/sulphur/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sulphur/subscriptions","organizations_url":"https://api.github.com/users/sulphur/orgs","repos_url":"https://api.github.com/users/sulphur/repos","events_url":"https://api.github.com/users/sulphur/events{/privacy}","received_events_url":"https://api.github.com/users/sulphur/received_events","type":"User","site_admin":false},"created_at":"2015-07-30T08:50:42Z","updated_at":"2015-07-30T08:50:42Z","body":"In my case i have 3 brokres on my local machine. I'm running a consumers in tasks processes that are using KafkaEx.fetch\r\n\r\nFor instance if i do this in console : \r\n```iex(14)> KafkaEx.fetch(\"k_users\",2,[offset: 775, worker_name: :pr])```\r\n\r\nI got following error if leader for partition 2 is down :\r\n\r\n```\r\n** (exit) exited in: GenServer.call(:pr, {:fetch, \"k_users\", 2, 775, 10, 1, 1000000, true}, 5000)\r\n    ** (EXIT) an exception was raised:\r\n        ** (FunctionClauseError) no function clause matching in KafkaEx.Protocol.Fetch.parse_response/1\r\n            (kafka_ex) lib/kafka_ex/protocol/fetch.ex:18: KafkaEx.Protocol.Fetch.parse_response(nil)\r\n            (kafka_ex) lib/kafka_ex/server.ex:241: KafkaEx.Server.fetch/8\r\n            (kafka_ex) lib/kafka_ex/server.ex:50: KafkaEx.Server.handle_call/3\r\n            (stdlib) gen_server.erl:607: :gen_server.try_handle_call/4\r\n            (stdlib) gen_server.erl:639: :gen_server.handle_msg/5\r\n            (stdlib) proc_lib.erl:237: :proc_lib.init_p_do_apply/3\r\n    (elixir) lib/gen_server.ex:356: GenServer.call/3\r\niex(14)> \r\n10:46:31.111 [error] GenServer :pr terminating\r\nLast message: {:fetch, \"k_users\", 2, 775, 10, 1, 1000000, true}\r\nState: %KafkaEx.Server{brokers: [%KafkaEx.Protocol.Metadata.Broker{host: \"192.168.1.88\", node_id: 1, port: 9092, socket: #Port<0.9732>}, %KafkaEx.Protocol.Metadata.Broker{host: \"192.168.1.88\", node_id: 2, port: 9093, socket: #Port<0.9731>}, %KafkaEx.Protocol.Metadata.Broker{host: \"192.168.1.88\", node_id: 3, port: 9094, socket: #Port<0.9730>}], consumer_group: \"kafka_ex\", consumer_metadata: %KafkaEx.Protocol.ConsumerMetadata.Response{coordinator_host: \"192.168.1.88\", coordinator_id: 2, coordinator_port: 9093, error_code: 0}, correlation_id: 16, event_pid: nil, metadata: %KafkaEx.Protocol.Metadata.Response{brokers: [%KafkaEx.Protocol.Metadata.Broker{host: \"192.168.1.88\", node_id: 3, port: 9094, socket: nil}, %KafkaEx.Protocol.Metadata.Broker{host: \"192.168.1.88\", node_id: 2, port: 9093, socket: nil}, %KafkaEx.Protocol.Metadata.Broker{host: \"192.168.1.88\", node_id: 1, port: 9092, socket: nil}], topic_metadatas: [%KafkaEx.Protocol.Metadata.TopicMetadata{error_code: 0, partition_metadatas: [%KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 18, replicas: [3, 2, 1]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 0, replicas: [3, 2, 1]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 9, replicas: [3, 1, 2]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 28, replicas: [1, 2, 3]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 41, replicas: [2, 3, 1]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 36, replicas: [3, 2, 1]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 31, replicas: [1, 3, 2]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 19, replicas: [1, 3, 2]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 14, replicas: [2, 1, 3]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 32, replicas: [2, 1, 3]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 22, replicas: [1, 2, 3]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 27, replicas: [3, 1, 2]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 34, replicas: [1, 2, 3]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 2, replicas: [2, 1, 3]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 39, replicas: [3, 1, 2]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 25, replicas: [1, 3, 2]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 20, replicas: [2, 1, 3]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 30, replicas: [3, 2, 1]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 47, replicas: [2, 3, 1]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 43, replicas: [1, 3, 2]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 6, replicas: [3, 2, 1]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 38, replicas: [2, 1, 3]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 46, replicas: [1, 2, 3]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 1, replicas: [1, 3, 2]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 33, replicas: [3, 1, 2]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 29, replicas: [2, 3, 1]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 13, replicas: [1, 3, 2]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 24, replicas: [3, 2, 1]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 4, replicas: [1, 2, 3]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 26, replicas: [2, 1, 3]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 42, replicas: [3, 2, 1]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 37, replicas: [1, 3, 2]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 17, replicas: [2, 3, 1]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 12, replicas: [3, 2, 1]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 49, replicas: [1, 3, ...]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 7, replicas: [1, ...]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 44, ...}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, ...], leader: 1, ...}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [...], ...}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, ...}, %KafkaEx.Protocol.Metadata.PartitionMetadata{...}, ...], topic: \"__consumer_offsets\"}, %KafkaEx.Protocol.Metadata.TopicMetadata{error_code: 0, partition_metadatas: [%KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [1, 2], leader: 1, partition_id: 0, replicas: [1, 3, 2]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 2, replicas: [3, 2, 1]}, %KafkaEx.Protocol.Metadata.PartitionMetadata{error_code: 0, isrs: [2, 1], leader: 2, partition_id: 1, replicas: [2, 1, 3]}], topic: \"k_users\"}]}}\r\n** (exit) an exception was raised:\r\n    ** (FunctionClauseError) no function clause matching in KafkaEx.Protocol.Fetch.parse_response/1\r\n        (kafka_ex) lib/kafka_ex/protocol/fetch.ex:18: KafkaEx.Protocol.Fetch.parse_response(nil)\r\n        (kafka_ex) lib/kafka_ex/server.ex:241: KafkaEx.Server.fetch/8\r\n        (kafka_ex) lib/kafka_ex/server.ex:50: KafkaEx.Server.handle_call/3\r\n        (stdlib) gen_server.erl:607: :gen_server.try_handle_call/4\r\n        (stdlib) gen_server.erl:639: :gen_server.handle_msg/5\r\n        (stdlib) proc_lib.erl:237: :proc_lib.init_p_do_apply/3\r\n\r\n10:46:31.111 [info]  Application kafka_ex exited: shutdown\r\n ``` \r\n\r\nAnd after that kafka_en is shut down and i cannot restart is (i need to re-launch iex) \r\n"}}