# Log stash pulling messages from kafka

**URL:** <https://discuss.elastic.co/t/log-stash-pulling-messages-from-kafka/49510>\
**Category:** Elasticsearch\
**Created:** [May 9, 2016, 5:37am UTC](https://discuss.elastic.co/t/log-stash-pulling-messages-from-kafka/49510 "2016-05-09T05:37:40Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![bbhandari0121](https://avatars.discourse-cdn.com/v4/letter/b/f0a364/32.png) [@bbhandari0121](https://discuss.elastic.co/u/bbhandari0121)\
**Post date:** [May 9, 2016, 5:37am UTC](https://discuss.elastic.co/t/log-stash-pulling-messages-from-kafka/49510/1 "2016-05-09T05:37:40Z")

</div>

Hi team,  
I have 3 kafka brokers, and 2 topics each have 15 partitions. Also, i have got the cluster of ES with 5 nodes, and each of them have 2 log stash config running to pull messages from 2 kafka topices. But i see its not reading the messages from all the partitions, and also, i see there is a delay in writing the message to the ES.  
input {  
kafka {  
zk\_connect =\> ':2181,:2181,:2181'  
topic\_id =\> 'collectdmetric'  
consumer\_threads =\> 3  
}  
}

output {  
if [type] == 'collectd' {  
elasticsearch {  
hosts =\> [":9200",":9200",":9200"]  
manage\_template =\> false  
index =\> "collectdmetrics-%{+YYYY.MM.dd}"  
}  
}  
}

input {  
kafka {  
zk\_connect =\> ':2181,:2181,:2181'  
topic\_id =\> 'tomcat'  
consumer\_threads =\> 3  
}  
}

output {  
if [type] == 'tomcat' {  
elasticsearch {  
hosts =\> [":9200",":9200",":9200"]  
manage\_template =\> false  
index =\> "tomcat-%{+YYYY.MM.dd}"  
}  
}  
}  
health status index pri rep docs.count docs.deleted store.size pri.store.size  
green open collectdmetrics-2016.05.03 5 1 98612060 0 37gb 18.5gb  
green open tomcat-2016.05.03 5 1 9861200 0 6gb 3gb  
Can some one please guide to to the right directions ?

---

<div class="post-metadata">

**Author:** ![mainec](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/mainec/32/5557_2.png) [@mainec](https://discuss.elastic.co/u/mainec)\
**Post date:** [May 10, 2016, 8:56am UTC](https://discuss.elastic.co/t/log-stash-pulling-messages-from-kafka/49510/2 "2016-05-10T08:56:55Z")

</div>

I think this question might be better answered by people over in the Logstash group.

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [July 5, 2017, 10:52pm UTC](https://discuss.elastic.co/t/log-stash-pulling-messages-from-kafka/49510/3 "2017-07-05T22:52:50Z")

</div>