Proposal - Support the possibility ta handle exceptions in poll() loop

[jrask]

Hi,
and thanks for an awesome project!

I have previously posted a message on elastic discuss forum without any response so I will try here instead, post is here.

Our current issue is that when running kafka with SASL_SSL we can get "weird" errors during cluster restart and logstash does not handle any exceptions. I would like to propose a config change so we can actually handle exceptions and decide whether to continue or crash. However, it is very likely that there are different scenarios that we should be able to handle.

I would like to suggest two enhancements

1. Default always retry on RetriableException unless explicitly disabled.

2. Support adding an array of exceptions that should result in retry.

Something like the following would simply invoke poll() again when this error occurs.
retry_on => ["org.apache.kafka.common.errors.SaslAuthenticationException","some_other_error"] sleep_ms_before_retry => 1000

In my opinion there is rarely any danger in letting a consumer retry "indefinitely".

If this is of interest I can of course send a PR.

[colinsurprenant]

The logstash philosophy to to always retry upon transient errors until that error condition is resolved.
Note that we also have identified cases of exceptions not correctly handled in kafka related to elastic/logstash#11603