# Kfaka not writing to elasticsearch using logstash

**URL:** <https://discuss.elastic.co/t/kfaka-not-writing-to-elasticsearch-using-logstash/77109>\
**Category:** Elasticsearch\
**Created:** [March 2, 2017, 8:19am UTC](https://discuss.elastic.co/t/kfaka-not-writing-to-elasticsearch-using-logstash/77109 "2017-03-02T08:19:19Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![tshebin](https://avatars.discourse-cdn.com/v4/letter/t/df705f/32.png) [@tshebin](https://discuss.elastic.co/u/tshebin)\
**Post date:** [March 2, 2017, 8:19am UTC](https://discuss.elastic.co/t/kfaka-not-writing-to-elasticsearch-using-logstash/77109/1 "2017-03-02T08:19:19Z")

</div>

# Here is the producer's config file of logstash:

input {  
#file {

# path =\> "/var/log/messages"

# start\_position =\> "beginning"

#}  
stdin { }  
}

output {  
kafka {  
bootstrap\_servers =\> "kfaka-server-01:9092,kfaka-server-02:9092,kafka-server-03:9092"  
topic\_id =\> "topic01-2017-03-02"  
compression\_type =\> "snappy"  
}  
#stdout { codec =\> rubydebug }  
}

# consumer's config file of logstash:

input {  
kafka {  
bootstrap\_servers =\> "kafka-server-01:2181,kfaka-server-02:2181,kfaka-server-03:2181"  
topics =\> "topic01-2017-03-02"  
#codec =\> plain  
#reset\_beginning =\> false  
#consumer\_threads =\> 5  
#decorate\_events =\> true  
}  
#stdin { }  
}

output {  
elasticsearch {  
hosts =\> ["es-server-01:9200","es-server-02:9200"]  
index =\> "index01-2017-03-02"  
}  
stdout { codec =\> rubydebug }  
}

# When I input sth from stdin in the producer's side, I can check out the messages from kafka using [kfaka-console-consume.sh](http://kfaka-console-consume.sh) at the same time. That means the first logstash config file is correct, right?

But I cannot get any messages from elasticsearch, including the index.  
ps: i can get messages when i change the 2nd config file's input to stdin.  
can anyone help?

# my soft version:

elasticsearch 5.2.1  
logstash 5.2.1  
kfaka 2.12-0.10.2.0

---

<div class="post-metadata">

**Author:** ![wenpos](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/wenpos/32/15326_2.png) [@wenpos](https://discuss.elastic.co/u/wenpos)\
**Post date:** [March 3, 2017, 8:08am UTC](https://discuss.elastic.co/t/kfaka-not-writing-to-elasticsearch-using-logstash/77109/2 "2017-03-03T08:08:20Z")

</div>

Some suggestion as fllows:  
**1. topics is array type like this:**  
topics =\> ["topic01-2017-03-02"]

**2. set a new group\_id, if there exists another logstash instance consumer with the same group id, and the offset was consumed to the latest:**  
group\_id =\> "new\_group"

**3. set auto\_offset\_reset to earliest to start a new offset, or set to none to make sure if the previous offset would be found or occupied (exception would be thrown):**  
auto\_offset\_reset =\> "earliest"  
or  
auto\_offset\_reset =\> "none"

hoping this helps  
--Fanfan

---

<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:** [March 31, 2017, 8:08am UTC](https://discuss.elastic.co/t/kfaka-not-writing-to-elasticsearch-using-logstash/77109/3 "2017-03-31T08:08:45Z")

</div>

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