# Should Kafka Input handle Exceptions during poll()?

**URL:** <https://discuss.elastic.co/t/should-kafka-input-handle-exceptions-during-poll/134210>\
**Category:** Logstash\
**Created:** [June 1, 2018, 11:42am UTC](https://discuss.elastic.co/t/should-kafka-input-handle-exceptions-during-poll/134210 "2018-06-01T11:42:31Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![JohanRask](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/johanrask/32/22713_2.png) [@JohanRask](https://discuss.elastic.co/u/JohanRask)\
**Post date:** [June 1, 2018, 11:42am UTC](https://discuss.elastic.co/t/should-kafka-input-handle-exceptions-during-poll/134210/1 "2018-06-01T11:42:32Z")

</div>

Hi,

Kafka clients producer and consumer does good job in making sure that any errors are not propagated up to logstash. However, some errors can slip through and especially during a cluster restart.

Our problem is that with SASL\_SSL enabled, the consumer recieves an authentication exception when it reconnects. This error occurs in brokers as well but they retry until it works.

I would suggest adding a begin/rescue around the poll() method and catch ApiExceptions there. I do not see any problem to simply continue even if an error persists instead of terminating the pipeline.

This approach solves our problem since "under-the-hoods", KafkaConsumer will resume connection to brokers successfully.

```
#Begin before poll()
begin 

  records = consumer.poll(poll_timeout_ms) # throws Exception

  for record in records do
    codec_instance.decode(record.value.to_s) do |event|
      decorate(event)
      if @decorate_events
        event.set("[@metadata][kafka][topic]", record.topic)
        event.set("[@metadata][kafka][consumer_group]", @group_id)
        event.set("[@metadata][kafka][partition]", record.partition)
        event.set("[@metadata][kafka][offset]", record.offset)
        event.set("[@metadata][kafka][key]", record.key)
        event.set("[@metadata][kafka][timestamp]", record.timestamp)
      end
      logstash_queue << event
    end
  end
  # Manual offset commit
  if @enable_auto_commit == "false"
    consumer.commitSync
  end

# Rescue on ApiException
rescue org.apache.kafka.common.errors.ApiException => ae
  logger.warn("Exception during poll occured")
  # Perhaps sleep a few seconds before continue?
end
```

---

<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:** [June 29, 2018, 11:42am UTC](https://discuss.elastic.co/t/should-kafka-input-handle-exceptions-during-poll/134210/2 "2018-06-29T11:42:33Z")

</div>

Closed on June 29, 2018, 11:42am UTC

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