# Logstash can not consume messages from kafka **URL:** \ **Category:** Logstash\ **Created:** [May 14, 2015, 4:59am UTC](https://discuss.elastic.co/t/logstash-can-not-consume-messages-from-kafka/664 "2015-05-14T04:59:29Z")\ **Posts on this page:** 6\ **Page:** 1
**Author:** ![wivernband](https://avatars.discourse-cdn.com/v4/letter/w/a698b9/32.png) [@wivernband](https://discuss.elastic.co/u/wivernband)\ **Post date:** [May 14, 2015, 4:59am UTC](https://discuss.elastic.co/t/logstash-can-not-consume-messages-from-kafka/664/1 "2015-05-14T04:59:29Z")
We configured logstash to use "kafka" as input and "elasticsearch" as output. But it seems that the logstash instance can not consume the messages from kafka. Kindly check the logstash version and its configuration as below: > [root@elk logstash-1.5]# logstash --version > logstash 1.5.0.rc2 > [root@elk logstash-1.5]# cat /etc/logstash/central.conf > input { > kafka { > zk\_connect =\> "hdp4:2181" > topic\_id =\> "test" > } > } > output { > elasticsearch { > cluster =\> "logstash" > } > } As you can see, the connection with zookeeper has established: > [root@elk logstash-1.5]# netstat -an | grep 2181 > tcp 0 0 ::ffff:10.28.0.233:54260 ::ffff:10.80.3.234:2181 ESTABLISHED > [root@elk logstash-1.5]# ping hdp4 > PING hdp4 (10.80.3.234) 56(84) bytes of data. > 64 bytes from hdp4 (10.80.3.234): icmp\_seq=1 ttl=63 time=0.729 ms > 64 bytes from hdp4 (10.80.3.234): icmp\_seq=2 ttl=63 time=1.08 ms And we can also assure that there are messages in kafka can be consumed: > [root@hdp6 bin]# ./kafka-console-consumer.sh --zookeeper 10.80.3.234:2181 --topic test --from-beginning > May 13 14:25:54 lek-aix61 auth|security:err|error sshd[6619256]: error: Could not load host key: /etc/ssh/ssh\_host\_ecdsa\_key > May 13 14:25:56 lek-aix61 auth|security:info sshd[6619256]: Accepted password for root from 10.28.0.233 port 58526 ssh2 > May 13 14:29:20 lek-aix61 auth|security:info sshd[6619256]: Received disconnect from 10.28.0.233: 11: disconnected by user Logged into the zookeeper server, and we found that there was no "offset" sub-directory generated for logstash: > [zk: localhost:2181(CONNECTED) 31] ls /consumers/console-consumer-28776 > [offsets, owners, ids] > [zk: localhost:2181(CONNECTED) 32] ls /consumers/logstash > [owners, ids] Thanks a lot for your support in advance. ---
**Author:** ![wivernband](https://avatars.discourse-cdn.com/v4/letter/w/a698b9/32.png) [@wivernband](https://discuss.elastic.co/u/wivernband)\ **Post date:** [May 14, 2015, 7:54am UTC](https://discuss.elastic.co/t/logstash-can-not-consume-messages-from-kafka/664/2 "2015-05-14T07:54:41Z")
Noticed the change logs in the new released Logstash 1.5.0 RC4: > - Kafka: > - Merged @joekiller's plugin to Logstash to get events from Kafka (#1472) > - Added support for whitelisting and blacklisting topics in the input. Will use the new release to have test. ---
**Author:** ![wivernband](https://avatars.discourse-cdn.com/v4/letter/w/a698b9/32.png) [@wivernband](https://discuss.elastic.co/u/wivernband)\ **Post date:** [May 15, 2015, 6:03am UTC](https://discuss.elastic.co/t/logstash-can-not-consume-messages-from-kafka/664/3 "2015-05-15T06:03:38Z")
Anybody can help on the error below: ``` {:timestamp=>"2015-05-15T13:54:57.043000+0800", :message=>"Failed to flush outgoing items", :outgoing_count=>2, :exception=>#, :backtrace=>["/opt/logstash-1.5/vendor/bundle/jruby/1.9/gems/logstash-output-elasticsearch-0.2.4-java/lib/logstash/outputs/elasticsearch.rb:464:in `flush'", "/opt/logstash-1.5/vendor/bundle/jruby/1.9/gems/logstash-output-elasticsearch-0.2.4-java/lib/logstash/outputs/elasticsearch.rb:462:in `flush'", "/opt/logstash-1.5/vendor/bundle/jruby/1.9/gems/logstash-output-elasticsearch-0.2.4-java/lib/logstash/outputs/elasticsearch.rb:460:in `flush'", "/opt/logstash-1.5/vendor/bundle/jruby/1.9/gems/stud-0.0.19/lib/stud/buffer.rb:219:in `buffer_flush'", "org/jruby/RubyHash.java:1341:in `each'", "/opt/logstash-1.5/vendor/bundle/jruby/1.9/gems/stud-0.0.19/lib/stud/buffer.rb:216:in `buffer_flush'", "/opt/logstash-1.5/vendor/bundle/jruby/1.9/gems/stud-0.0.19/lib/stud/buffer.rb:112:in `buffer_initialize'", "org/jruby/RubyKernel.java:1507:in `loop'", "/opt/logstash-1.5/vendor/bundle/jruby/1.9/gems/stud-0.0.19/lib/stud/buffer.rb:110:in `buffer_initialize'"], :level=>:warn, :file=>"stud/buffer.rb", :line=>"231", :method=>"buffer_flush"} ``` Tks! ---
**Author:** ![Suresh\_Lasantha](https://avatars.discourse-cdn.com/v4/letter/s/82dd89/32.png) [@Suresh\_Lasantha](https://discuss.elastic.co/u/Suresh_Lasantha)\ **Post date:** [June 23, 2015, 8:39am UTC](https://discuss.elastic.co/t/logstash-can-not-consume-messages-from-kafka/664/4 "2015-06-23T08:39:06Z")
I think you've got the same problem described [here](https://github.com/logstash-plugins/logstash-output-elasticsearch/issues/154). update the logstash-output-elasticsearch plugin to the latest version. ---
**Author:** ![Vtamins](https://avatars.discourse-cdn.com/v4/letter/v/97f17d/32.png) [@Vtamins](https://discuss.elastic.co/u/Vtamins)\ **Post date:** [May 16, 2016, 10:03am UTC](https://discuss.elastic.co/t/logstash-can-not-consume-messages-from-kafka/664/5 "2016-05-16T10:03:44Z")
I have a same problem like this。 [2016-05-16 17:56:56,147] INFO Got user-level KeeperException when processing sessionid:0x154b859cdaa0018 type:create cxid:0x14 zxid:0x24a txntype:-1 reqpath:n/a Error Path:/consumers/logstash/owners/mytopic Error:KeeperErrorCode = NoNode for /consumers/logstash/owners/mytopic (org.apache.zookeeper.server.PrepRequestProcessor) Can you tell me how do you deal with this？ thanks ---
**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:57am UTC](https://discuss.elastic.co/t/logstash-can-not-consume-messages-from-kafka/664/6 "2017-07-06T04:57:30Z")