# Kafka Input Tutorial for Logstash 5.1.1

**URL:** <https://discuss.elastic.co/t/kafka-input-tutorial-for-logstash-5-1-1/70762>\
**Category:** Logstash\
**Created:** [January 6, 2017, 9:31am UTC](https://discuss.elastic.co/t/kafka-input-tutorial-for-logstash-5-1-1/70762 "2017-01-06T09:31:55Z")\
**Posts on this page:** 7\
**Page:** 1

<div class="post-metadata">

**Author:** ![gringo](https://avatars.discourse-cdn.com/v4/letter/g/f19dbf/32.png) [@gringo](https://discuss.elastic.co/u/gringo)\
**Post date:** [January 6, 2017, 9:31am UTC](https://discuss.elastic.co/t/kafka-input-tutorial-for-logstash-5-1-1/70762/1 "2017-01-06T09:31:55Z")

</div>

Is there any third party documentation on how to use the Kafka plugin. Below is my logstash script

input  
{  
kafka  
{  
bootstrap\_servers =\> ["192.168.31.55:9092,192.168.31.54:9092"]  
topics =\> "test-beat33"  
}  
}

filter  
{

}  
output  
{  
stdout { codec =\> rubydebug }  
}

---

<div class="post-metadata">

**Author:** ![gringo](https://avatars.discourse-cdn.com/v4/letter/g/f19dbf/32.png) [@gringo](https://discuss.elastic.co/u/gringo)\
**Post date:** [January 6, 2017, 9:32am UTC](https://discuss.elastic.co/t/kafka-input-tutorial-for-logstash-5-1-1/70762/2 "2017-01-06T09:32:37Z")

</div>

Below is what I see on the console

c:\test\logstash-script\>..\logstash-5.1.1\bin\logstash -f test.conf  
Could not find log4j2 configuration at path /Users/A/test/logstash-5.1.1/config/log4j2.properties. Using default config which logs to console  
16:51:46.992 [[main]-pipeline-manager] INFO logstash.pipeline - Starting pipeline {"id"=\>"main", "pipeline.workers"=\>4, "pipeline.batch.size"=\>125, "pipeline.batch.delay"=\>5, "pipeline.max\_inflight"=\>500}  
16:51:47.009 [[main]-pipeline-manager] INFO logstash.pipeline - Pipeline main started  
16:51:47.106 [[main]\<kafka] 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 = [192.168.31.55:9092, 192.168.31.54:9092]  
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

16:51:47.191 [Api Webserver] INFO logstash.agent - Successfully started Logstash API endpoint {:port=\>9600}  
16:51:47.249 [[main]\<kafka] 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 = [192.168.31.55:9092, 192.168.31.54:9092]  
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

---

<div class="post-metadata">

**Author:** ![gringo](https://avatars.discourse-cdn.com/v4/letter/g/f19dbf/32.png) [@gringo](https://discuss.elastic.co/u/gringo)\
**Post date:** [January 6, 2017, 9:32am UTC](https://discuss.elastic.co/t/kafka-input-tutorial-for-logstash-5-1-1/70762/3 "2017-01-06T09:32:48Z")

</div>

16:51:47.339 [[main]\<kafka] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version : 0.10.0.1  
16:51:47.341 [[main]\<kafka] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId : a7a17cdec9eaa6c5  
16:51:47.513 [Ruby-0-Thread-16: C:/Users/A/test/logstash-5.1.1/vendor/bundle/jruby/1.9/gems/logstash-input-kafka-5.1.0/lib/logstash/inputs/kafka.rb:225] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - Discovered coordinator 192.168.31.54:9092 (id: 2147483646 rack: null) for group logstash.  
16:51:47.519 [Ruby-0-Thread-16: C:/Users/A/test/logstash-5.1.1/vendor/bundle/jruby/1.9/gems/logstash-input-kafka-5.1.0/lib/logstash/inputs/kafka.rb:225] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - Revoking previously assigned partitions [] for group logstash  
16:51:47.523 [Ruby-0-Thread-16: C:/Users/A/test/logstash-5.1.1/vendor/bundle/jruby/1.9/gems/logstash-input-kafka-5.1.0/lib/logstash/inputs/kafka.rb:225] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - (Re-)joining group logstash  
16:51:47.556 [Ruby-0-Thread-16: C:/Users/A/test/logstash-5.1.1/vendor/bundle/jruby/1.9/gems/logstash-input-kafka-5.1.0/lib/logstash/inputs/kafka.rb:225] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - Successfully joined group logstash with generation 5  
16:51:47.561 [Ruby-0-Thread-16: C:/Users/A/test/logstash-5.1.1/vendor/bundle/jruby/1.9/gems/logstash-input-kafka-5.1.0/lib/logstash/inputs/kafka.rb:225] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - Setting newly assigned partitions [test-beat33-0] for group logstash

Despite seeing so many messages I cannot see any logs in the console. When I check the Kafka server using the kafka-console-consumer I can see my logs.

Is there any third party tutorial or documentation that I can use? Is there something wrong with logstash or there are some secret information on usage that is not disclosed to the public

---

<div class="post-metadata">

**Author:** ![sandeepkanabar](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/sandeepkanabar/32/79399_2.png) [@sandeepkanabar](https://discuss.elastic.co/u/sandeepkanabar)\
**Post date:** [January 6, 2017, 3:26pm UTC](https://discuss.elastic.co/t/kafka-input-tutorial-for-logstash-5-1-1/70762/4 "2017-01-06T15:26:43Z")

</div>

> [@gringo](#):
>
> Despite seeing so many messages I cannot see any logs in the console

Which console are you referring to? Your logstash console?

> Could not find log4j2 configuration at path /Users/A/test/logstash-5.1.1/config/log4j2.properties. Using default config which logs to console

You can try to start logstash as:  
`..\logstash-5.1.1\bin\logstash -f test.conf --path.settings=dir_name_where_logstash.yml_is_located --config.debug --log.level debug` after which you can see the logs in logstash logs directory.

---

<div class="post-metadata">

**Author:** ![gringo](https://avatars.discourse-cdn.com/v4/letter/g/f19dbf/32.png) [@gringo](https://discuss.elastic.co/u/gringo)\
**Post date:** [January 9, 2017, 8:56am UTC](https://discuss.elastic.co/t/kafka-input-tutorial-for-logstash-5-1-1/70762/5 "2017-01-09T08:56:09Z")

</div>

I know what is the problem. All data that arrived in kafka before logstash has started will automatically ignored by logstash. only data that arrives after logstash has already started will be processed by logstash.

Looks like default logstash configuration will result in data loss should logstash server goes down or need to be restarted in a production environment.

Is there anyway to solve this problem? This problem will defeat the purpose of using Kafka

---

<div class="post-metadata">

**Author:** ![sandeepkanabar](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/sandeepkanabar/32/79399_2.png) [@sandeepkanabar](https://discuss.elastic.co/u/sandeepkanabar)\
**Post date:** [January 9, 2017, 5:30pm UTC](https://discuss.elastic.co/t/kafka-input-tutorial-for-logstash-5-1-1/70762/6 "2017-01-09T17:30:21Z")

</div>

See - [https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#plugins-inputs-kafka-enable\_auto\_commit](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#plugins-inputs-kafka-enable_auto_commit). By default this is set to true and so Kafka knows how much the logstash consumer group has read and knows the offset position. Does this answer your question?

---

<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:** [February 6, 2017, 5:30pm UTC](https://discuss.elastic.co/t/kafka-input-tutorial-for-logstash-5-1-1/70762/7 "2017-02-06T17:30:32Z")

</div>

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