Logstash cannot consume from Kafka

Question

My structure is like this: Logfiles > Filebeat > Kafka > Logstash > Elasticsearch > Kibana

But I am stuck at the Kafka to Logstash part.

First, Filebeat can produce the message to Kafka and i can check it by using:

kafka-console-consumer.bat --zookeeper localhost:2181 --topic {topicName} --from-beginning

And it can also be consumed by command:

kafka-console-consumer.bat --zookeeper localhost:2181 --topic {topicName}

However, when I try to use logstash to consume the topic, nothing can be retrieved and Zookeeper keep throwing:

2017-11-13 16:11:59,205 [myid:] - INFO [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxnFactory@215] - Accepted socket connection from /127.0.0.1:53201

2017-11-13 16:11:59,207 [myid:] - WARN [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:21 81:NIOServerCnxn@383] - Exception causing close of session 0x0: null

2017-11-13 16:11:59,209 [myid:] - INFO [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxn@1040] - Closed socket connection for client /127.0.0.1:53201 (no session established for client)

Also, the port number in this statement keep increasing"Accepted socket connection from /127.0.0.1:53201". Is it scanning the port?

Below are some background. Version:

zookeeper-3.4.11

kafka_2.10-0.10.2.1

logstash-5.4.0

(Plugin)logstash-input-kafka (5.1.6)

And here are my config:

Zoo.cfg (zookeeper)

tickTime=2000
initLimit=10
syncLimit=5
clientPort=2181

server.properties (Kafka)

listeners=PLAINTEXT://localhost:9092
zookeeper.connect=127.0.0.1:2181

logstash.config (logstash)

input {
    kafka {
        bootstrap_servers => "localhost:2181"
        topics => "test"
    }   
}

I guess the problem is due to the config between zookeeper and kafka but i have no idea how to figure it out. Pls help.

Alex Ott Alex Ott · Accepted Answer · 2017-11-13T09:29:52

You need to use localhost:9092 for configuration parameter bootstrap_servers - you're pointing Logstash to Zookeeper, but you need to point to Kafka itself.

Another error is that topics requires an array, so it should be ["test"] instead of "test"

Logstash cannot consume from Kafka

1 Answers