# Unable to get Logstash to read from kafka input

**URL:** https://discuss.elastic.co/t/unable-to-get-logstash-to-read-from-kafka-input/82090
**Category:** Logstash
**Created:** [April 12, 2017, 5:34am UTC](https://discuss.elastic.co/t/unable-to-get-logstash-to-read-from-kafka-input/82090 "2017-04-12T05:34:33Z")
**Posts on this page:** 5
**Page:** 1

<div class="post-metadata">

### Author: ![vramakrishnan](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/vramakrishnan/32/31750_2.png) [@vramakrishnan](https://discuss.elastic.co/u/vramakrishnan)
#### Post date: [April 12, 2017, 5:34am UTC](https://discuss.elastic.co/t/unable-to-get-logstash-to-read-from-kafka-input/82090/1 "2017-04-12T05:34:33Z")

</div>

In my setup I have Filebeats -\> Kafka -\> Logstash -\> ElasticSearch.

I can see filebeats producing logs into Kakfa-topic and I am also able to read logs from kafka topic using kafka-consumer-client. But I do not see logs read by log stash for the topic configured. The log stash logs don't show any errors and nothing gets printed in stdout or in the log stash logs for the configs below.

Any help is appreciated.

Version:

Fillebeat :Filebeat-5.2.0  
Kafka: kafka\_2.11- **0.10.2.0**  
Logstash: logstash 5.1.1

**Filebeat config:**

output.kafka:  
hosts: ["10.1.1.1:9092"]  
topic: 'csos\_services\_file'  
compression: gzip  
max\_message\_bytes: 1000000

output.console:  
pretty: true

**Logstash config:**

input {  
kafka {  
bootstrap\_servers =\> ["10.1.1.1:9092"]  
type =\> "kafka-input"  
topics =\> "csos\_services\_file"  
}

filter {  
}

output {  
stdout { codec =\> json }  
}

output {  
elasticsearch {  
hosts =\> ["10.6.0.4:9200"]  
user =\> "es\_admin"  
password =\> "XXXX"  
index =\> "csos\_services\_index.%{+YYYY.MM.dd}"  
}  
}

**Logstash Logs:**

[2017-04-12T05:14:47,049][INFO][org.apache.kafka.clients.consumer.ConsumerConfig] ConsumerConfig values:  
metric.reporters = []  
[metadata.max.age.ms](http://metadata.max.age.ms) = 300000  
partition.assignment.strategy = [org.apache.kafka.clients.consumer.RangeAssignor]  
[reconnect.backoff.ms](http://reconnect.backoff.ms) = 50  
sasl.kerberos.ticket.renew.window.factor = 0.8  
max.partition.fetch.bytes = 1048576  
bootstrap.servers = [10.1.1.1:2181]  
ssl.keystore.type = JKS  
enable.auto.commit = true  
sasl.mechanism = GSSAPI  
interceptor.classes = null  
exclude.internal.topics = true  
ssl.truststore.password = null  
[client.id](http://client.id) = logstash  
ssl.endpoint.identification.algorithm = null  
max.poll.records = 2147483647  
check.crcs = true  
[request.timeout.ms](http://request.timeout.ms) = 40000  
[heartbeat.interval.ms](http://heartbeat.interval.ms) = 3000  
[auto.commit.interval.ms](http://auto.commit.interval.ms) = 5000  
receive.buffer.bytes = 65536  
ssl.truststore.type = JKS  
ssl.truststore.location = null  
ssl.keystore.password = null  
fetch.min.bytes = 1  
send.buffer.bytes = 131072  
value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer  
[group.id](http://group.id) = logstash  
[retry.backoff.ms](http://retry.backoff.ms) = 100  
sasl.kerberos.kinit.cmd = /usr/bin/kinit  
[sasl.kerberos.service.name](http://sasl.kerberos.service.name) = null  
sasl.kerberos.ticket.renew.jitter = 0.05  
ssl.trustmanager.algorithm = PKIX  
ssl.key.password = null  
[fetch.max.wait.ms](http://fetch.max.wait.ms) = 500  
sasl.kerberos.min.time.before.relogin = 60000  
[connections.max.idle.ms](http://connections.max.idle.ms) = 540000  
[session.timeout.ms](http://session.timeout.ms) = 30000  
metrics.num.samples = 2  
key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer  
ssl.protocol = TLS  
ssl.provider = null  
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]  
ssl.keystore.location = null  
ssl.cipher.suites = null  
security.protocol = PLAINTEXT  
ssl.keymanager.algorithm = SunX509  
[metrics.sample.window.ms](http://metrics.sample.window.ms) = 30000  
auto.offset.reset = latest

**[2017-04-12T05:14:47,161][INFO][org.apache.kafka.common.utils.AppInfoParser] Kafka version : 0.10.0.1**  
[2017-04-12T05:14:47,161][INFO][org.apache.kafka.common.utils.AppInfoParser] Kafka commitId : a7a17cdec9eaa6c5

[2017-04-12T05:14:47,490][INFO][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=\>{:removed=\>[], :added=\>["http://~hidden~:~hidden~@10.6.0.4:9200"]}}  
[2017-04-12T05:14:47,492][INFO][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:url=\>#\<URI::HTTP:0x39f67862 URL:http://~hidden~:~hidden~@10.6.0.4:9200\>, :healthcheck\_path=\>"/"}  
[2017-04-12T05:14:47,739][WARN][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=\>#\<URI::HTTP:0x39f67862 URL:http://~hidden~:~hidden~@10.6.0.4:9200\>}  
[2017-04-12T05:14:47,762][INFO][logstash.outputs.elasticsearch] Using mapping template from {:path=\>nil}  
[2017-04-12T05:14:47,829][INFO][logstash.outputs.elasticsearch] Attempting to install template {:manage\_template=\>{"template"=\>"logstash-_", "settings"=\>{"index.refresh\_interval"=\>"5s"}, "mappings"=\>{"default"=\>{"\_all"=\>{"enabled"=\>true, "omit\_norms"=\>true}, "dynamic\_templates"=\>[{"message\_field"=\>{"path\_match"=\>"message", "match\_mapping\_type"=\>"string", "mapping"=\>{"type"=\>"string", "index"=\>"analyzed", "omit\_norms"=\>true, "fielddata"=\>{"format"=\>"disabled"}}}}, {"string\_fields"=\>{"match"=\>"_", "match\_mapping\_type"=\>"string", "mapping"=\>{"type"=\>"string", "index"=\>"analyzed", "omit\_norms"=\>true, "fielddata"=\>{"format"=\>"disabled"}, "fields"=\>{"raw"=\>{"type"=\>"string", "index"=\>"not\_analyzed", "doc\_values"=\>true, "ignore\_above"=\>256}}}}}, {"float\_fields"=\>{"match"=\>"_", "match\_mapping\_type"=\>"float", "mapping"=\>{"type"=\>"float", "doc\_values"=\>true}}}, {"double\_fields"=\>{"match"=\>"_", "match\_mapping\_type"=\>"double", "mapping"=\>{"type"=\>"double", "doc\_values"=\>true}}}, {"byte\_fields"=\>{"match"=\>"_", "match\_mapping\_type"=\>"byte", "mapping"=\>{"type"=\>"byte", "doc\_values"=\>true}}}, {"short\_fields"=\>{"match"=\>"_", "match\_mapping\_type"=\>"short", "mapping"=\>{"type"=\>"short", "doc\_values"=\>true}}}, {"integer\_fields"=\>{"match"=\>"_", "match\_mapping\_type"=\>"integer", "mapping"=\>{"type"=\>"integer", "doc\_values"=\>true}}}, {"long\_fields"=\>{"match"=\>"_", "match\_mapping\_type"=\>"long", "mapping"=\>{"type"=\>"long", "doc\_values"=\>true}}}, {"date\_fields"=\>{"match"=\>"_", "match\_mapping\_type"=\>"date", "mapping"=\>{"type"=\>"date", "doc\_values"=\>true}}}, {"geo\_point\_fields"=\>{"match"=\>"_", "match\_mapping\_type"=\>"geo\_point", "mapping"=\>{"type"=\>"geo\_point", "doc\_values"=\>true}}}], "properties"=\>{"@timestamp"=\>{"type"=\>"date", "doc\_values"=\>true}, "@version"=\>{"type"=\>"string", "index"=\>"not\_analyzed", "doc\_values"=\>true}, "geoip"=\>{"type"=\>"object", "dynamic"=\>true, "properties"=\>{"ip"=\>{"type"=\>"ip", "doc\_values"=\>true}, "location"=\>{"type"=\>"geo\_point", "doc\_values"=\>true}, "latitude"=\>{"type"=\>"float", "doc\_values"=\>true}, "longitude"=\>{"type"=\>"float", "doc\_values"=\>true}}}}}}}}  
[2017-04-12T05:14:47,837][INFO][logstash.outputs.elasticsearch] New Elasticsearch output {:class=\>"LogStash::Outputs::ElasticSearch", :hosts=\>["10.6.0.4:9200"]}  
[2017-04-12T05:14:47,850][INFO][logstash.pipeline] Starting pipeline {"id"=\>"main", "pipeline.workers"=\>1, "pipeline.batch.size"=\>125, "pipeline.batch.delay"=\>5, "pipeline.max\_inflight"=\>125}  
[2017-04-12T05:14:47,857][INFO][logstash.pipeline] Pipeline main started  
[2017-04-12T05:14:47,935][INFO][logstash.agent] Successfully started Logstash API endpoint {:port=\>9600}

---

<div class="post-metadata">

### Author: ![vramakrishnan](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/vramakrishnan/32/31750_2.png) [@vramakrishnan](https://discuss.elastic.co/u/vramakrishnan)
#### Post date: [April 14, 2017, 5:22am UTC](https://discuss.elastic.co/t/unable-to-get-logstash-to-read-from-kafka-input/82090/2 "2017-04-14T05:22:55Z")

</div>

Any help is appreciated on this one/

---

<div class="post-metadata">

### Author: ![riddhijit\_roy](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/riddhijit_roy/32/19049_2.png) [@riddhijit\_roy](https://discuss.elastic.co/u/riddhijit_roy)
#### Post date: [April 14, 2017, 7:17am UTC](https://discuss.elastic.co/t/unable-to-get-logstash-to-read-from-kafka-input/82090/3 "2017-04-14T07:17:40Z")

</div>

> [@vramakrishnan](#):
>
> input { kafka { bootstrap\_servers =\> ["10.1.1.1:9092"] type =\> "kafka-input" topics =\> "csos\_services\_file" }

Try adding a group\_id to your input plugin.  
input {  
kafka {  
bootstrap\_servers =\> ["10.1.1.1:9092"]  
type =\> "kafka-input"  
topics =\> "csos\_services\_file"  
auto\_offset\_reset =\> "earliest"  
group\_id =\> "Dexter"  
}

---

<div class="post-metadata">

### Author: ![vramakrishnan](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/vramakrishnan/32/31750_2.png) [@vramakrishnan](https://discuss.elastic.co/u/vramakrishnan)
#### Post date: [April 14, 2017, 4:11pm UTC](https://discuss.elastic.co/t/unable-to-get-logstash-to-read-from-kafka-input/82090/4 "2017-04-14T16:11:17Z")

</div>

> [@riddhijit\_roy](#):
>
> auto\_offset\_reset =\> "earliest"group\_id =\> "Dexter"

Thanks for the suggestion, But it didn't work with group\_id as well.

Is kafka input supported with logstash 5.1.1 ? And no debug logs seen on logstash

Fillebeat :Filebeat-5.2.0  
Kafka: kafka\_2.11-0.10.2.0  
Logstash: logstash 5.1.1

---

<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: [May 12, 2017, 4:20pm UTC](https://discuss.elastic.co/t/unable-to-get-logstash-to-read-from-kafka-input/82090/5 "2017-05-12T16:20:47Z")

</div>

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