# Logstash with multiple kafka inputs

**URL:** <https://discuss.elastic.co/t/logstash-with-multiple-kafka-inputs/57646>\
**Category:** Logstash\
**Created:** [August 10, 2016, 3:11am UTC](https://discuss.elastic.co/t/logstash-with-multiple-kafka-inputs/57646 "2016-08-10T03:11:12Z")\
**Posts on this page:** 6\
**Page:** 1

<div class="post-metadata">

**Author:** ![abhi\_chinchole](https://avatars.discourse-cdn.com/v4/letter/a/d6d6ee/32.png) [@abhi\_chinchole](https://discuss.elastic.co/u/abhi_chinchole)\
**Post date:** [August 10, 2016, 3:11am UTC](https://discuss.elastic.co/t/logstash-with-multiple-kafka-inputs/57646/1 "2016-08-10T03:11:12Z")

</div>

I am trying to filter kafka events from multiple topics, but once all events from one topic has been filtered logstash is not able to fetch events from the other kafka topic. I am using topics with 3 partitions and 2 replications Here is my logstash confi

input {  
kafka{  
auto\_offset\_reset =\> "smallest"  
consumer\_id =\> "logstashConsumer1"  
topic\_id =\> "unprocessed\_log1"  
zk\_connect=\>"192.42.79.67:2181,192.41.85.48:2181,192.10.13.14:2181"  
type =\> "kafka\_type\_1"  
}  
kafka{  
auto\_offset\_reset =\> "smallest"  
consumer\_id =\> "logstashConsumer1"  
topic\_id =\> "unprocessed\_log2"  
zk\_connect =\> "192.42.79.67:2181,192.41.85.48:2181,192.10.13.14:2181"  
type =\> "kafka\_type\_2"  
}  
}  
filter{  
if [type] == "kafka\_type\_1"{  
csv {  
separator=\>" "  
source =\> "data"  
}  
}  
if [type] == "kafka\_type\_2"{  
csv {  
separator =\> " "  
source =\> "data"  
}  
}  
}  
output{  
stdout{ codec=\>rubydebug{metadata =\> true }}  
}

---

<div class="post-metadata">

**Author:** ![warkolm](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/warkolm/32/39224_2.png) [@warkolm](https://discuss.elastic.co/u/warkolm)\
**Post date:** [August 10, 2016, 3:12am UTC](https://discuss.elastic.co/t/logstash-with-multiple-kafka-inputs/57646/2 "2016-08-10T03:12:09Z")

</div>

Please edit your subject so that you are not answering the entire question there.

---

<div class="post-metadata">

**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:** [August 10, 2016, 2:08pm UTC](https://discuss.elastic.co/t/logstash-with-multiple-kafka-inputs/57646/3 "2016-08-10T14:08:06Z")

</div>

With two kafka inputs the messages are processed in parallel. Are you just at the end of the data?

---

<div class="post-metadata">

**Author:** ![abhi\_chinchole](https://avatars.discourse-cdn.com/v4/letter/a/d6d6ee/32.png) [@abhi\_chinchole](https://discuss.elastic.co/u/abhi_chinchole)\
**Post date:** [August 12, 2016, 6:58am UTC](https://discuss.elastic.co/t/logstash-with-multiple-kafka-inputs/57646/4 "2016-08-12T06:58:30Z")

</div>

With above configurations data of only one topic was processed. My expectation was to process all the data from both topics. Now I have used different consumer\_id of both inputs and its working perfectly fine.

---

<div class="post-metadata">

**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:** [August 15, 2016, 4:10pm UTC](https://discuss.elastic.co/t/logstash-with-multiple-kafka-inputs/57646/5 "2016-08-15T16:10:18Z")

</div>

> [@abhi\_chinchole](#):
>
> Now I have used different consumer\_id of both inputs and its working perfectly fine.

Curious. I would expect each to be able to use the same as long as the topic is different. Thanks for the update!

---

<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 6, 2017, 4:43am UTC](https://discuss.elastic.co/t/logstash-with-multiple-kafka-inputs/57646/6 "2017-07-06T04:43:29Z")

</div>