# Configuring "consumer\_threads" for a Kafka input with multiple Logstash instances **URL:** \ **Category:** Logstash\ **Tags:** elastic-stack-monitoring\ **Created:** [July 17, 2024, 12:02pm UTC](https://discuss.elastic.co/t/configuring-consumer-threads-for-a-kafka-input-with-multiple-logstash-instances/363279 "2024-07-17T12:02:59Z")\ **Posts on this page:** 5\ **Page:** 1
**Author:** ![mmfsilva](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/mmfsilva/32/136099_2.png) [@mmfsilva](https://discuss.elastic.co/u/mmfsilva)\ **Post date:** [July 17, 2024, 12:02pm UTC](https://discuss.elastic.co/t/configuring-consumer-threads-for-a-kafka-input-with-multiple-logstash-instances/363279/1 "2024-07-17T12:02:59Z")
I'm analysing the way a logging stack has been configured and am facing a particular issue. We have 2 Logstash instances. In those two instances, we have two pipelines which are consuming from Kafka. See the `input` section of the pipeline below: ```auto input { kafka { bootstrap_servers => ["broker1:9092,broker2:9092,broker3:9092"] topics => "topic-name" security_protocol => "SSL" ssl_truststore_location => "/path/to/truststore/truststore.jks" ssl_truststore_password => "XYZ" client_id => "client-id" consumer_threads => "10" } } ``` The above pipeline configuration is the same for both Logstash instances. Regarding Kafka - this particular topic has a total of 8 partitions. _ **Issue #1** _ - According to [this thread](https://discuss.elastic.co/t/multiple-logstash-reading-from-a-single-kafka-topic/27727) we have a total of 20 consumer threads across both Logstash instances, but since only 8 partitions exist for this consumer group, that means there's a total of 12 `consumer_threads` which are idle at all times between both Logstash servers. For this scenario, the correct configuration would be to have 4 `consumer_threads` on each Logstash instance. Is this conclusion correct? _ **Issue #2** _ - Checking the number of workers on Logstash, for this particular pipeline I can see the following output of running `curl -XGET -k http://127.0.0.1:9600/_node/pipelines?pretty`: ```auto "pipeline-name" : { "ephemeral_id" : "553bc152-e29f-4c45-88f7-f5b77ff5a5b2", "hash" : "751c0c4bd2060bb70793328544279246418c135f3354596212ebb40247b3d5ff", "workers" : 16, "batch_size" : 125, "batch_delay" : 50, "config_reload_automatic" : false, "config_reload_interval" : 3000000000, "dead_letter_queue_enabled" : false } ``` Considering we have 16 `workers` on each Logstash instance, does that directly correlate with the `consume_threads` set on the `pipeline.conf`, meaning each instance would have 16 workers \* 10 consumer threads = 160 consumer threads on each Logstash instance? **For context** : the end-goal here is to improve the _Consumer lag_ which currently is extremely high and is leading to a 3-5 hour delay in logs showing up on OpenSearch where Logstash is currently sending the logs to. Our idea was to increase the number of partitions for the Kafka topics with a higher number of messages so that we can also directly increase the Logstash `consumer_threads` consuming the topics (again, according the shared thread above these consumer threads have a 1:1 ration with partitions on Kafka) and hopefully decrease the lag. Thank you ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [July 17, 2024, 12:03pm UTC](https://discuss.elastic.co/t/configuring-consumer-threads-for-a-kafka-input-with-multiple-logstash-instances/363279/2 "2024-07-17T12:03:00Z")
OpenSearch/OpenDistro are AWS run products and differ from the original Elasticsearch and Kibana products that Elastic builds and maintains. You may need to contact them directly for further assistance. See [What is OpenSearch and the OpenSearch Dashboard? | Elastic](https://www.elastic.co/elasticsearch/opensearch) for more details. (This is an automated response from your friendly Elastic bot. Please report this post if you have any suggestions or concerns :elasticheart: ) ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [July 17, 2024, 1:18pm UTC](https://discuss.elastic.co/t/configuring-consumer-threads-for-a-kafka-input-with-multiple-logstash-instances/363279/3 "2024-07-17T13:18:44Z")
> [@mmfsilva](#): > > Considering we have 16 `workers` on each Logstash instance, does that directly correlate with the `consume_threads` set on the `pipeline.conf`, meaning each instance would have 16 workers \* 10 consumer threads = 160 consumer threads on each Logstash instance? No, it does not. You have 10 threads running the input, and 16 threads running the filters and outputs. ---
**Author:** ![mmfsilva](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/mmfsilva/32/136099_2.png) [@mmfsilva](https://discuss.elastic.co/u/mmfsilva)\ **Post date:** [July 17, 2024, 2:55pm UTC](https://discuss.elastic.co/t/configuring-consumer-threads-for-a-kafka-input-with-multiple-logstash-instances/363279/4 "2024-07-17T14:55:39Z")
So this means we can take the `workers` out of the equation when considering the number of `consumer_threads` per logstash instance, per Kafka topic replica. Thank you! ---
**Author:** ![leandrojmp](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/leandrojmp/32/107231_2.png) [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)\ **Post date:** [July 17, 2024, 3:08pm UTC](https://discuss.elastic.co/t/configuring-consumer-threads-for-a-kafka-input-with-multiple-logstash-instances/363279/5 "2024-07-17T15:08:28Z")
> [@mmfsilva](#): > > Our idea was to increase the number of partitions for the Kafka topics with a higher number of messages so that we can also directly increase the Logstash `consumer_threads` consuming the topics (again, according the shared thread above these consumer threads have a 1:1 ration with partitions on Kafka) and hopefully decrease the lag. You should have as many `consumer_threads` as the number of partitions in your topic, if you add another logstash instance, then you need to recalculate this. For example, with 8 partitions and 2 instances, you may have 4 consumer threads each, as you mentioned in the _Issue #1_ part. Another thing that you can do to improve the input is to change the default value of [`max_poll_records`](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#plugins-inputs-kafka-max_poll_records), from `500` to `1000` for example. And your output will also impact in the lag size as if the destination can not process the data fast enough, it will tell logstash to backoff a little, so you caldo also increase the value of `pipeline.batch_size` from the default `125` to another number, start with the double for example.