Add ability to conditionally retry messages with KafkaListener's retry error stragegy

[MRyanEvans]

Desired behaviour

A listener with the retry error strategy always retries the message; however, there are some scerarios where a retry is not necessary or desirable.

For example, there are two types of deserialisation errors:

• Deterministic errors where the message is not of the expected type or its schema does not exist (and is unlikely to ever exist) in the schema registry.
• Transient errors, such as network issues when talking to the schema registry.

For these cases it would be desireable to skip the first one and retry the second.

This is particularly helpful where we would not want to miss valid messages and have retryCount = Integer.MAX_VALUE, where we'd have to wait ages until our application eventually skips the message with the current behaviour.

Workaround

The only way that I've been able to achieve this desired behaviour is to disable the retry mechanism and implement my own KafkaListenerExceptionHandler which seeks back to the message or forwards to the next depending on the exception thrown. The downside is that I lose the ability to easily add retry delays and exponential back-offs.

Suggested fix

My PR reorders the seek and exception handling so that any seeking the exception handler decides to do is honoured.

Additional notes

I am happy to define the seek forward/backward behaviour in my own exception handler; however it might be simpler if you were to open up a similar hander that just needs to return 'retry' or 'skip' and have the seeking done in your code.

TL;DR

Basically, I want my custom exception handler with the retry strategy to be able to skip messages.

@KafkaListener(
    errorStrategy = @ErrorStrategy(value = RETRY_ON_ERROR, handleAllExceptions = true)
)
static class FooConsumer implements KafkaListenerExceptionHandler {
    @Topic("foo")
    void handleMessage(ConsumerRecord record) {
        doAThing(record);
    }

    @Override
    void handle(KafkaListenerException exception) {
        if(message is bad) {
            exception.kafkaConsumer.seek(new TopicPartition(record.topic(), record.partition()), record.offset() + 1);
        }
    }
}