# Kafka consumer issue **URL:** \ **Category:** Logstash\ **Created:** [March 18, 2016, 1:39pm UTC](https://discuss.elastic.co/t/kafka-consumer-issue/44840 "2016-03-18T13:39:16Z")\ **Posts on this page:** 4\ **Page:** 1
**Author:** ![ravitandur](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/ravitandur/32/4568_2.png) [@ravitandur](https://discuss.elastic.co/u/ravitandur)\ **Post date:** [March 18, 2016, 1:39pm UTC](https://discuss.elastic.co/t/kafka-consumer-issue/44840/1 "2016-03-18T13:39:16Z")
We are using logstash kafka input plugin to process the data from kafka. We are running 3 log stash instances with following config: input { heartbeat { interval =\> 60 type =\> "heartbeat" add\_field =\> { "[app.name](http://app.name)" =\> "LSKTOES" } } kafka { zk\_connect =\> "zk-host:2181" topic\_id =\> "central-logging" consumer\_threads =\> 4 } } output { elasticsearch { host =\> ["ls-host-1","ls-host-2","ls-host-3"] port =\> "9200" protocol =\> "http" codec =\> json workers =\> 4 } } We are running Kafka on node [one process] with 12 partations. So with 3 log stash agents with each having 4 consumer threads should fetch the data fro 12 partitions. But this is not working. One of the logstash consumer, is not consuming any message and in logs i am getting these statements continuously log4j, [2016-03-18T09:32:22.904] INFO: kafka.consumer.ConsumerFetcherManager: [ConsumerFetcherManager-1458306025713] Added fetcher for partitions ArrayBuffer() log4j, [2016-03-18T09:32:23.115] INFO: kafka.utils.VerifiableProperties: Verifying properties log4j, [2016-03-18T09:32:23.115] INFO: kafka.utils.VerifiableProperties: Property [client.id](http://client.id) is overridden to logstash log4j, [2016-03-18T09:32:23.115] INFO: kafka.utils.VerifiableProperties: Property metadata.broker.list is overridden to 10.87.164.36:9092 log4j, [2016-03-18T09:32:23.115] INFO: kafka.utils.VerifiableProperties: Property [request.timeout.ms](http://request.timeout.ms) is overridden to 30000 log4j, [2016-03-18T09:32:23.115] INFO: kafka.client.ClientUtils$: Fetching metadata from broker id:1,host:10.87.164.36,port:9092 with correlation id 8652 for 1 topic(s) Set(central-logging) log4j, [2016-03-18T09:32:23.116] INFO: kafka.producer.SyncProducer: Connected to 10.87.164.36:9092 for producing log4j, [2016-03-18T09:32:23.117] INFO: kafka.producer.SyncProducer: Disconnecting from 10.87.164.36:9092 log4j, [2016-03-18T09:32:23.117] INFO: kafka.consumer.ConsumerFetcherManager: [ConsumerFetcherManager-1458306025713] Added fetcher for partitions ArrayBuffer() log4j, [2016-03-18T09:32:23.336] INFO: kafka.utils.VerifiableProperties: Verifying properties log4j, [2016-03-18T09:32:23.337] INFO: kafka.utils.VerifiableProperties: Property [client.id](http://client.id) is overridden to logstash log4j, [2016-03-18T09:32:23.337] INFO: kafka.utils.VerifiableProperties: Property metadata.broker.list is overridden to 10.87.164.36:9092 log4j, [2016-03-18T09:32:23.337] INFO: kafka.utils.VerifiableProperties: Property [request.timeout.ms](http://request.timeout.ms) is overridden to 30000 log4j, [2016-03-18T09:32:23.337] INFO: kafka.client.ClientUtils$: Fetching metadata from broker id:1,host:10.87.164.36,port:9092 with correlation id 8653 for 1 topic(s) Set(central-logging) log4j, [2016-03-18T09:32:23.337] INFO: kafka.producer.SyncProducer: Connected to 10.87.164.36:9092 for producing log4j, [2016-03-18T09:32:23.338] INFO: kafka.producer.SyncProducer: Disconnecting from 10.87.164.36:9092 log4j, [2016-03-18T09:32:23.338] INFO: kafka.consumer.ConsumerFetcherManager: [ConsumerFetcherManager-1458306025713] Added fetcher for partitions ArrayBuffer() When I look the kafka server log, I am seeing the below logs which indicates server is closing the sockets continuously on consumers. Not very sure why connection is getting closed continuously. [2016-03-18 09:37:56,147] INFO Closing socket connection to /10.100.124.47. (kafka.network.Processor) [2016-03-18 09:37:56,234] INFO Closing socket connection to /10.100.124.49. (kafka.network.Processor) [2016-03-18 09:37:56,338] INFO Closing socket connection to /10.100.124.48. (kafka.network.Processor) The IPs mentioned in the above logs are 3 logstash consumer IPs. Any idea what might be wrong? ---
**Author:** ![Joe\_Lawson](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/joe_lawson/32/3390_2.png) [@Joe\_Lawson](https://discuss.elastic.co/u/Joe_Lawson)\ **Post date:** [March 24, 2016, 6:13pm UTC](https://discuss.elastic.co/t/kafka-consumer-issue/44840/2 "2016-03-24T18:13:12Z")
I'm guessing you are getting an error in the Kafka pipeline and logstash is restarting over and over. Try running in --debug and set on the kafka input `consumer_restart_on_error => false` ([https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#plugins-inputs-kafka-consumer\_restart\_on\_error](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#plugins-inputs-kafka-consumer_restart_on_error)) ---
**Author:** ![ravitandur](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/ravitandur/32/4568_2.png) [@ravitandur](https://discuss.elastic.co/u/ravitandur)\ **Post date:** [March 31, 2016, 7:56am UTC](https://discuss.elastic.co/t/kafka-consumer-issue/44840/3 "2016-03-31T07:56:29Z")
I tried with the option mentioned, but still seeing the same issue. Any further ideas please? ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [July 6, 2017, 5:04am UTC](https://discuss.elastic.co/t/kafka-consumer-issue/44840/4 "2017-07-06T05:04:35Z")