Better error handling support for reactive functions

I'm using the new functional approach with RabbitMQ and strive for fully reactive message handling.
When using a reactive consumer, any error will break the connection.
What I'd like instead is, if any exception/Flux.error() happens, the RabbitMQ should retry/publish message to dlq.

Considering the following example:

@Bean fun addAccount(dispatcher: ...): Consumer<Flux<Message<String>>> = Consumer { it.flatMap { mono { dispatcher.handle(it) } } .onErrorContinue { ex, _ -> StaticMessageHeaderAccessor.getAcknowledgmentCallback(...)?.noAutoAck() }.subscribe() }

I understand, that I have to handle errors manually, maybe sending a manual ack/nack or requeue requests. When using non reactive logic this is provided out of the box. It would be nice if I could achieve the same behavior with an error handler in the reactive chain. The above example doesn't work as there is no ACKNOWLEDGMENT_CALLBACK in the message (that's added in non reactive case).

Any solution that achieves "retry on error" behaviour would be nice.

Start with the reactive consumer example and the reported absence of an acknowledgment callback, then compare its error behavior with the non-reactive RabbitMQ path. Done means an exception or Flux.error() can trigger retry or dead-letter publication without breaking the connection, with the behavior covered by an appropriate test.