In what version(s) of Spring for Apache Kafka are you seeing this issue?

4.0.3

Describe the bug

Maybe this is by design, but I couldn't find concrete documentation.

When configuring a container error handler together with an async listener, the handler is not called.

To Reproduce

Configure a Kafka listener that returns a failed CompletableFuture.

Expected behavior

The handling for completable futures takes KafkaListenerErrorHandler into account, but if that handler is not configured, any exception is just thrown and never handled.

From what I see in the code, I think the final future returned by whenComplete (from the previous link) should be propagated all the way to the listener container, which would need to check that specific case and probably handle the future to call invokeErrorHandler when there's an unhandled async failure, similar to what it does currently in the catch blocks for sync failures.

After that I'm not sure what's best. I imagine the listener container would have to keep track of those futures, wait for all of them to complete, and fail the batch if any of them failed even after invokeErrorHandler?