# Logstash-output-kafka stops producing data without error if topic is not valid

**URL:** <https://discuss.elastic.co/t/logstash-output-kafka-stops-producing-data-without-error-if-topic-is-not-valid/54167>\
**Category:** Logstash\
**Created:** [June 28, 2016, 1:35pm UTC](https://discuss.elastic.co/t/logstash-output-kafka-stops-producing-data-without-error-if-topic-is-not-valid/54167 "2016-06-28T13:35:09Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![fld](https://avatars.discourse-cdn.com/v4/letter/f/65b543/32.png) [@fld](https://discuss.elastic.co/u/fld)\
**Post date:** [June 28, 2016, 1:35pm UTC](https://discuss.elastic.co/t/logstash-output-kafka-stops-producing-data-without-error-if-topic-is-not-valid/54167/1 "2016-06-28T13:35:09Z")

</div>

When using kafka output plugin, if your topic is extracted using a grok filter and if for some reason the filter fails (grokparsefailure), then the output plugin is blocked and no explicit error is produced.

**Steps to reproduce:**

1. Get a kafka broker running on localhost:9092

2. Create a `sample.log` file containing the following lines:

> Matching test: 1  
> Matching test: 2  
> Not matching test: 3  
> Matching test: 4  
> Matching test: 5

1. Use the following logstash configuration to push the data to the local kafka broker:

> input {  
> file {  
> path =\> "/home/user/sample.log"  
> start\_position =\> beginning  
> }  
> }
> 
> filter {  
> grok {  
> match =\> { "message" =\> "^Matching %{WORD:topic}: %{NUMBER:line}.\*" }  
> }  
> }
> 
> output {  
> stdout { codec =\> rubydebug }  
> kafka {  
> batch\_size =\> 1  
> bootstrap\_servers =\> "127.0.0.1:9092"  
> topic\_id =\> "%{topic}"  
> }  
> }

1. Read the content of the topic using kafka console consumer:  
`$KAFKA_INSTALL_DIR/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning`

**Observed result:**

Only the first two lines are pushed to the topic:

> {"message":"Matching test: 1","@version":"1","@timestamp":"2016-06-28T13:23:14.276Z","path":"/home/user/sample.log","host":"fabien1.dev","topic":"test","line":"1"}  
> {"message":"Matching test: 2","@version":"1","@timestamp":"2016-06-28T13:23:14.632Z","path":"/home/user/sample.log","host":"fabien1.dev","topic":"test","line":"2"}

The first missing event is:

> {  
> "message" =\> "Not matching test: 3",  
> "@version" =\> "1",  
> "@timestamp" =\> "2016-06-28T13:30:37.454Z",  
> "path" =\> "/home/user/sample.log",  
> "host" =\> "host1.dev",  
> "tags" =\> [  
> [0] "\_grokparsefailure"  
> ]  
> }

Logstash has read, according to the stdout output, all the lines (the last ones are stuck in some buffers) and is looping pushing some data containing "%{topic}" to kafka. Nothing else will happen, nothing in the logs

Logstash: 2.1.1 to 2.3.3  
Kafka: 2.10-0.10.0.0  
OS: Linux debian wheezy

For me this is an issue: logstash should at least detect it and produce an error message.  
A simple fix in the configuration is to check that the parsing for the topic is not failing.

---

<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, 4:50am UTC](https://discuss.elastic.co/t/logstash-output-kafka-stops-producing-data-without-error-if-topic-is-not-valid/54167/2 "2017-07-06T04:50:39Z")

</div>