Spring cloud stream dlq processing with spring cloud function for rabbitmq

Question

I have read the spring cloud stream binder reference document which mentioned DLQ processing using @RabbitListener. https://docs.spring.io/spring-cloud-stream-binder-rabbit/docs/3.0.10.RELEASE/reference/html/spring-cloud-stream-binder-rabbit.html#rabbit-dlq-processing

Can we achieve the same via Spring cloud function like we can do the same for consumers? Like

@Bean
    public Consumer<Message> dlqprocess(DLQProcess dlqprocess) {
        return t -> dlqprocess.do(t);
    }

I am not sure whether we can do this or not. If this allows what are the other configuration we have to do?

Answer 1

If you aim is to requeue failed messages, the function can just throw exceptions as described in docs.

Furthermore, if you need more fine-grained control about send and requeued messages you can use StreamBrdidge. Here you need to explicitly define DLQ binding in the configuration file:

spring.cloud.stream.bindings.myDlq-out-0.destination=DLX
spring.cloud.stream.rabbit.bindings.myDlq-out-0.producer.exchangeType=direct
spring.cloud.stream.rabbit.bindings.myDlq-out-0.producer.routingKeyExpression='myDestination.myGroup'
spring.cloud.stream.source=myDlq

Finally, the function controls whether to send and requeue the message:

@Bean
public Consumer<Message> process(StreamBridge streamBridge) {
    return t -> {
        // ....
        if(republish) streamBridge.send("myDlq-out-0", t);
        if(sendToDestination) streamBridge.send("myDestination-out-0", t);
        // ....
    };
}