Retry Mechanism Not Working with Reactive Types in Micronaut Kafka using @KafkaListener

[EnriqueWood]

Expected Behavior

When using the @KafkaListener annotation with @ErrorStrategy(value = ErrorStrategyValue.RETRY_ON_ERROR), it is expected that the listener will retry the specified number of times (retryCount) with a certain delay (retryDelay) between retries when an exception is thrown. This behavior should be consistent regardless of whether the method returns a reactive type (e.g., Mono) or a non-reactive type.

Actual Behaviour

The retry mechanism works as expected for methods that do not return a reactive type. However, for methods returning a reactive type, such as Mono, the retry mechanism does not activate upon encountering exceptions. Instead of retrying, the error is logged, and no further action is taken, which leads to message processing being halted without retries.

Steps To Reproduce

1. Configure a Micronaut application with micronaut-kafka integration.
2. Define a @KafkaListener class with two methods listening to two different topics: one returning a non-reactive type and another returning a reactive type (Mono), both configured with an @ErrorStrategy that specifies retry behavior.
3. Inject a message in each topic to simulate an error condition within both methods to trigger the retry mechanism.
4. Observe that for the non-reactive method, the retry mechanism works as expected, with retries being logged and attempted.
5. Observe that for the reactive method, the retry mechanism does not activate, and no retries are attempted after the initial error.

Code sample A WebhookServiceListener class is defined with two methods annotated with @Topic. One method, workingRetry, returns a non-reactive type and demonstrates the expected retry mechanism when an exception occurs. The other method, nonWorkingRetry, returns a reactive type (Mono

[jeremyg484]

Due to the default way Micronaut Kafka currently handles reactive return values (it optimistically continues the poll() loop and auto commits offsets without blocking), the only way I see that we can combine error strategies with reactive return values without breaking things would be to also require the @Blocking annotation on the reactive method. I was hoping that worked already, but it does not. I will put together a PR to fix that.