# Logstash does not consume Kafka

**URL:** <https://discuss.elastic.co/t/logstash-does-not-consume-kafka/281278>\
**Category:** Logstash\
**Tags:** docker\
**Created:** [August 13, 2021, 2:47am UTC](https://discuss.elastic.co/t/logstash-does-not-consume-kafka/281278 "2021-08-13T02:47:32Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![ktpktr0](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/ktpktr0/32/87157_2.png) [@ktpktr0](https://discuss.elastic.co/u/ktpktr0)\
**Post date:** [August 13, 2021, 2:47am UTC](https://discuss.elastic.co/t/logstash-does-not-consume-kafka/281278/1 "2021-08-13T02:47:32Z")

</div>

I created 3 logstash + 3 Kafka, and the initial size of each topic is 1. After a log grows greatly, the Kafka disk reaches 100%. After the disk is expanded, logstash does not consume Kafka. Only when Kafka is restarted can it consume some. After a stacked topic is cleared, logstash still does not consume Kafka, and more and more messages are stacked on a topic.

Later, I set the topic partition to 3 and configure the consumer with logstash\_ Threads = \> 3, but it doesn't solve the problem, and I see that only one of the three logstash is consuming.

logstash:

```auto
input {
    kafka {
       bootstrap_servers => "192.168.1.176:9092,192.168.1.178:9092,192.168.1.177:9092"
       client_id => "logstash175-2"
       auto_offset_reset => "latest"
       #auto_offset_reset => "earliest"
       reconnect_backoff_ms => 1000
       consumer_threads => 3
       topics => "docker"
       group_id => "docker"
       security_protocol => "SASL_PLAINTEXT"
       sasl_mechanism => "PLAIN"
       jaas_path => "/etc/logstash/kafka-client-jaas.conf"
    }
}

filter {
     json {
       skip_on_invalid_json => true
       source => "message"
   }
# ruby {
# code => "
# hash = event.to_hash
# hash.each do |k,v|
# if v == nil
# event.remove(k)
# end
# end
# "
# }
}

output {
   if [container][labels][service] {
       elasticsearch {
           hosts => ["192.168.1.74:9200","192.168.1.165:9200","192.168.1.166:9200"]
           user => "elastic"
           password => "SaOGLWtH4WnnJTgnu9vO"
           ssl => true
           ssl_certificate_verification => true
           cacert => "/etc/logstash/ssl/root.pem"
           manage_template => false
           index => "%{[container][labels][service]}-%{+YYYY.MM.dd}"
       }
   }

```

kafka:

```auto
/bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.178:9092 --command-config config/auth.conf --describe --group docker
[2021-08-13 10:39:04,465] WARN The configuration 'sasl.jaas.config' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig)

GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
docker docker 0 492973059 892839667 399866608 logstash173-2-0-a45bc193-3704-4bf3-99f3-abd14c29fc7d /192.168.1.173 logstash173-2-0
docker docker 1 11523898 15089801 3565903 logstash173-2-1-7ace1ed9-7bf6-4d28-af34-ef06186e9dd0 /192.168.1.173 logstash173-2-1
docker docker 2 11333226 15093179 3759953 logstash173-2-2-8f6e1668-1bfb-4683-af6f-da8f9fa1d3b1 /192.168.1.173 logstash173-2-2

```

---

<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:** [September 10, 2021, 2:48am UTC](https://discuss.elastic.co/t/logstash-does-not-consume-kafka/281278/2 "2021-09-10T02:48:16Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.