# Can't pull kafka messages into logstash and then to elasticsearch

**URL:** <https://discuss.elastic.co/t/cant-pull-kafka-messages-into-logstash-and-then-to-elasticsearch/46291>\
**Category:** Logstash\
**Created:** [April 4, 2016, 7:44pm UTC](https://discuss.elastic.co/t/cant-pull-kafka-messages-into-logstash-and-then-to-elasticsearch/46291 "2016-04-04T19:44:15Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![stecino](https://avatars.discourse-cdn.com/v4/letter/s/ea666f/32.png) [@stecino](https://discuss.elastic.co/u/stecino)\
**Post date:** [April 4, 2016, 7:44pm UTC](https://discuss.elastic.co/t/cant-pull-kafka-messages-into-logstash-and-then-to-elasticsearch/46291/1 "2016-04-04T19:44:15Z")

</div>

Hello,

I am getting this in my logstash log

> "{:timestamp=\>"2016-04-04T19:27:10.116000+0000", :message=\>"Flushing buffer at interval",  
> :instance=\>"#\<LogStash::Outputs::Elasticsearch::Buffer:0x635fc4ba operations\_mutex=#Mutex:0x6d9836a1,  
> max\_size=500, operations\_lock=#Java::JavaUtilConcurrentLocks::ReentrantLock:0x69a7afca,  
> submit\_proc=#Proc:0x314247a0/home/t/logstash-2.2.2/vendor/bundle/jruby/1.9/gems/logstash-output-elasticsearch-2.5.1-java/lib/logstash/outputs/elasticsearch/common.rb:57,  
> logger=#\<Cabin::Channel:0x7502b298 metrics=#\<Cabin::Metrics:0x4234fde metrics\_lock=#Mutex:0x24a36adf,  
> metrics={}, channel=#\<Cabin::Channel:0x7502b298 ...\>\>, subscriber\_lock=#Mutex:0x61124ce1,  
> level=:debug, subscribers={12906=\>#\<Cabin::Outputs::IO:0x1f274674 io=#\<File:/home/t/logstash-2.2.2/var/log/kafka\_logstash.log\>,  
> lock=#Mutex:0x20ed32dd\>}, data={}\>, last\_flush=2016-04-04 19:27:09 +0000, flush\_interval=1, stopping=#Concurrent::AtomicBoolean:0x1d435217,  
> buffer=, flush\_thread=#\<Thread:0x494e778a run\>\>", :interval=\>1, :level=\>:debug, :file=\>"logstash/outputs/elasticsearch/buffer.rb",  
> :line=\>"90", :method=\>"interval\_flush"} "

This is my config

input {

kafka {  
zk\_connect =\> "[blah.com:2181](http://blah.com:2181)"  
topic\_id =\> "f5-logs-wc1"  
codec =\> plain {  
format =\> "%{message}"  
}  
}  
}

filter {

```
    grok {
      break_on_match => false
      match => ["message", "%{SYSLOGTIMESTAMP} %{HOSTNAME:logsource} %{LOGLEVEL:severity_label} %{SYSLOGPROG}: %{GREEDYDATA:info}"]
    }
    mutate {
        gsub => [
            "severity_label", "err", "error",
            "severity_label", "info", "informational",
            "severity_label", "crit", "critical"
        ]

    remove_field => ["message","host"]
     
    }        

```

}

output {

```
     elasticsearch {
         hosts => ["blah1.com:9200", "blah2.com:9200"]
         index => "f5-%{+YYYY.MM.dd}"
                 }

  stdout { codec => rubydebug }

       }

```

I have verified that topic has data. So writing to kafka is working

---

<div class="post-metadata">

**Author:** ![Joe\_Lawson](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/joe_lawson/32/3390_2.png) [@Joe\_Lawson](https://discuss.elastic.co/u/Joe_Lawson)\
**Post date:** [April 13, 2016, 1:04am UTC](https://discuss.elastic.co/t/cant-pull-kafka-messages-into-logstash-and-then-to-elasticsearch/46291/2 "2016-04-13T01:04:57Z")

</div>

Try taking the filter out and only output to debug. Also run Logstash using the --debug, [https://www.elastic.co/guide/en/logstash/current/command-line-flags.html](https://www.elastic.co/guide/en/logstash/current/command-line-flags.html)

After it is up and running push messages to the topic.

---

<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:** [July 6, 2017, 5:02am UTC](https://discuss.elastic.co/t/cant-pull-kafka-messages-into-logstash-and-then-to-elasticsearch/46291/3 "2017-07-06T05:02:40Z")

</div>