CODEWITHSUNDEEP
javaspringspring-bootrabbitmqspring-cloud-stream
Vishal Mamidi
Vishal Mamidi

Reputation: 55

Messages not going to dead-letter queue on throwing exception

I have an application where I have to move the message to DLQ (dead-letter queue) on a particular case, but on the first message, the exception is thrown, but it ignores my max-attempts:2 value and I can see my exception message 4 times and the message has not gone to DLQ.

As mentioned in spring cloud stream docs, the message should go to DLQ on throwing an exception. When I try to send another message, I keep on getting a different exception in console logs until I restart the application to get first message again, but now the message goes to DLQ.

I tried using .onErrorContinue, in this case my application is not getting broken on first message, but messages are not going to DLQ.

EDIT:

There is an open incident on spring-cloud-stream repo for this issue https://github.com/spring-cloud/spring-cloud-stream/issues/1922

If anyone has any workaround to do retry, dead-letter-queue inside reactive function please provide an answer.

My application

    @Bean
    public Function<Flux<String>, Flux<?>> processor() {
        return PaymentDetailsFlux -> PaymentDetailsFlux
//                .retryWhen(Retry.backoff(3, Duration.ofMillis(100000)))
                .flatMap(
                        paymentDetails -> {
                            throw new RuntimeException("intentional");

                            /* extracting payload from input */
//                        Map<Object, Object> payload = new HashMap<>();
//                        try {
//                            ObjectMapper jsonMapper = new ObjectMapper();
//                            payload = jsonMapper.readValue(paymentDetails, Map.class);
//                        } catch (JsonProcessingException e) {
//                            log.error("error while serializing input payload: "+paymentDetails);
//                            e.printStackTrace();
//                        }
//                        String message_id = (String) payload.get("message_id");
//                        log.info("<"+message_id+">"+" Received message from payment-supplier");
//                        log.info("<"+message_id+">"+" message payload: "+paymentDetails);
//
//                        /* DLQ if rule not found */
//                        if(!isValidateMAP(payload)){
//                            throw new RuntimeException("intentional");
////                            return Flux.error(new RuntimeException("<"+message_id+"> "+errMessage));
//                        }
//                            return Mono.just(paymentDetails);
                        }
                );
//                .onErrorContinue((throwable, o) -> {
//                    log.error(throwable.getMessage());
//                });
    }

my application.yml

spring:
  application:
    name: payment-supplier
  profiles:
    active: dev
  cloud:
    function:
      definition: processor;ruleProcessor
    stream:
      bindings:
        processor-in-0:
          destination: processor-in     #input topic name
        processor-in-0.group: processor #queue name
        processor-out-0:
          destination: rule-in          #output topic name

        ruleProcessor-in-0:
          destination: ruleProcessor-in      #input topic name
        ruleProcessor-in-0.group: processor  #queue name
        #ruleProcessor-out-0:                #StreamBridge
      default-binder: rabbit

      #Defining DLQ - Dead Letter Queue
      rabbit:
        bindings:
          processor-in-0:
            consumer:
              auto-bind-dlq: true
              republish-to-dlq: true
              max-attempts: 2
              requeueRejected: true
              backOffInitialInterval: 900000
              backOffMaxInterval: 900000
            producer:
              autoBindDlq: true
              republishToDlq: true
              requeueRejected: true
              maxAttempts: 2
              backOffInitialInterval: 900000
              backOffMaxInterval: 900000


#Defining DLQ - Dead Letter Queue
#      rabbit:
        default:
          consumer:
            auto-bind-dlq: true
            republishToDlq : true
            maxAttempts : 3
            requeueRejected : true
#            backOffInitialInterval: 900000
#            backOffMaxInterval: 900000
          producer:
            auto-bind-dlq: true
            republishToDlq : true
            maxAttempts : 3
            requeueRejected : true
#            backOffInitialInterval: 900000
#            backOffMaxInterval: 900000

enter image description here First message

2021-10-18 13:12:40.704 ERROR 6704 --- [-in.processor-1] c.f.c.c.BeanFactoryAwareFunctionRegistry : Failed to invoke function 'processor'

java.lang.RuntimeException: intentional
    at com.gaic.paymentprocessor.service.PaymentProcessorService.lambda$processor$0(PaymentProcessorService.java:50) ~[classes/:na]
    springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.javaspringframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.lambda$onMessage$0(AmqpInboundChannelAdapter.java:374) ~[spring-integration-amqp-5.5.4.jar:5.5.4]
    at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329) ~[spring-retry-1.3.1.jar:na]
    at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225) ~[spring-retry-1.3.1.jar:na]
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:370) ~[spring-integration-amqp-5.5.4.jar:5.5.4]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(Aborg.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1498) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1195) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]

2021-10-18 13:12:40.704 ERROR 6704 --- [-in.processor-1] c.f.c.c.BeanFactoryAwareFunctionRegistry : Failed to convert output

java.lang.RuntimeException: intentional
    at com.gaic.paymentprocessor.service.PaymentProcessorService.lambda$processor$0(PaymentProcessorService.java:50) ~[classes/:na]
    springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1554) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1498) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:968) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:914) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1289) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1195) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]

2021-10-18 13:12:40.704 ERROR 6704 --- [-in.processor-1] onfiguration$FunctionToDestinationBinder : Failure was detected during execution of the reactive function 'processor'
2021-10-18 13:12:40.712 ERROR 6704 --- [-in.processor-1] reactor.core.publisher.Operators         : Operator called default onErrorDropped

reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.RuntimeException: intentional
Caused by: java.lang.RuntimeException: intentional
    at com.gaic.paymentprocessor.service.PaymentProcessorService.lambda$processor$0(PaymentProcessorService.java:50) ~[classes/:na]
    at reactor.core.publisher.FluxFlatMap$FlatMapMain.onNext(FluxFlatMap.java:386) ~[reactor-core-3.4.10.jar:3.4.10]
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120) ~[reactor-core-3.4.10.jar:3.4.10]
    at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:200) ~[reactor-core-3.4.10.jar:3.4.10]
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120) ~[reactor-core-3.4.10.jar:3.4.10]org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72) ~[spring-integration-core-5.5.4.jar:5.5.4]
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:317) ~[spring-integration-core-5.5.4.jar:5.5.4]
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:272) ~[spring-integration-core-5.5.4.jar:5.5.4]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187) ~[spring-messaging-5.3.10.jar:5.3.10]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166) ~[spring-messaging-5.3.10.jar:5.3.10]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47) ~[spring-messaging-5.3.10.jar:5.3.10]
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109) ~[spring-messaging-5.3.10.jar:5.3.10]
    at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:208) ~[spring-integration-core-5.5.4.jar:5.5.4]
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access$1300(AmqpInboundChannelAdapter.java:69) ~[spring-integration-amqp-5.5.4.jar:5.5.4]
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.lambda$onMessage$0(AmqpInboundChannelAdapter.java:374) ~[spring-integration-amqp-5.5.4.jar:5.5.4]
    at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329) ~[spring-retry-1.3.1.jar:na]
    at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225) ~[spring-retry-1.3.1.jar:na]
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:370) ~[spring-integration-amqp-5.5.4.jar:5.5.4]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1656) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1575) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1563) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1554) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1498) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:968) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:914) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1289) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1195) ~[spring-rabbit-2.3.10.jar:2.3.10]
    at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]

java.lang.RuntimeException: intentional
    at com.gaic.paymentprocessor.service.PaymentProcessorService.lambda$processor$0(PaymentProcessorService.java:50)
    at reactor.core.publisher.FluxFlatMap$FlatMapMain.onNext(FluxFlatMap.java:386)
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120)
    at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:200)
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120)
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120)
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120)
    at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:200)
    at reactor.core.publisher.UnicastManySinkNoBackpressure.tryEmitNext(UnicastManySinkNoBackpressure.java:120)
    at reactor.core.publisher.SinkManySerialized.tryEmitNext(SinkManySerialized.java:100)
    at org.springframework.integration.util.IntegrationReactiveUtils.lambda$adaptSubscribableChannelToPublisher$8(IntegrationReactiveUtils.java:142)
    at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:115)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:133)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:106)
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:317)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:272)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
    at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:208)
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access$1300(AmqpInboundChannelAdapter.java:69)
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.lambda$onMessage$0(AmqpInboundChannelAdapter.java:374)
    at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329)
    at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225)
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:370)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1656)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1575)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1563)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1554)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1498)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:968)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:914)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1289)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1195)
    at java.base/java.lang.Thread.run(Thread.java:834)

Second message logs

2021-10-18 12:57:25.422 ERROR 9928 --- [-in.processor-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'payment-supplier.processor-in-0'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[1330], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=processor-in, amqp_deliveryTag=2, deliveryAttempt=3, amqp_consumerQueue=processor-in.processor, amqp_redelivered=false, id=368d916d-af66-24c1-59c1-491d5dbc6dfa, amqp_consumerTag=amq.ctag-fibQI2iMkG5_giNWmMMwlA, sourceData=(Body:'[B@1d04b729(byte[1330])' MessageProperties [headers={}, contentLength=0, receivedDeliveryMode=PERSISTENT, redelivered=false, receivedExchange=processor-in, receivedRoutingKey=, deliveryTag=2, consumerTag=amq.ctag-fibQI2iMkG5_giNWmMMwlA, consumerQueue=processor-in.processor]), contentType=application/json, timestamp=1634542042374}], failedMessage=GenericMessage [payload=byte[1330], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=processor-in, amqp_deliveryTag=2, deliveryAttempt=3, amqp_consumerQueue=processor-in.processor, amqp_redelivered=false, id=368d916d-af66-24c1-59c1-491d5dbc6dfa, amqp_consumerTag=amq.ctag-fibQI2iMkG5_giNWmMMwlA, sourceData=(Body:'[B@1d04b729(byte[1330])' MessageProperties [headers={}, contentLength=0, receivedDeliveryMode=PERSISTENT, redelivered=false, receivedExchange=processor-in, receivedRoutingKey=, deliveryTag=2, consumerTag=amq.ctag-fibQI2iMkG5_giNWmMMwlA, consumerQueue=processor-in.processor]), contentType=application/json, timestamp=1634542042374}]
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:76)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:317)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:272)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
    at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:208)
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access$1300(AmqpInboundChannelAdapter.java:69)
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.lambda$onMessage$0(AmqpInboundChannelAdapter.java:374)
    at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329)
    at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225)
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:370)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1656)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1575)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1563)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1554)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1498)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:968)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:914)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1289)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1195)
    at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[1330], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=processor-in, amqp_deliveryTag=2, deliveryAttempt=3, amqp_consumerQueue=processor-in.processor, amqp_redelivered=false, id=368d916d-af66-24c1-59c1-491d5dbc6dfa, amqp_consumerTag=amq.ctag-fibQI2iMkG5_giNWmMMwlA, sourceData=(Body:'[B@1d04b729(byte[1330])' MessageProperties [headers={}, contentLength=0, receivedDeliveryMode=PERSISTENT, redelivered=false, receivedExchange=processor-in, receivedRoutingKey=, deliveryTag=2, consumerTag=amq.ctag-fibQI2iMkG5_giNWmMMwlA, consumerQueue=processor-in.processor]), contentType=application/json, timestamp=1634542042374}]
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:139)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:106)
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72)
    ... 23 more

Upvotes: 0

Views: 2123

Answers (1)

navid_gh
navid_gh

Reputation: 1883

I have answered the question in the Github (https://github.com/spring-cloud/spring-cloud-stream/issues/1922), here is the answer

Hi @vishalmamidi,

I have implemented my own version of DLQ by using StreamBridge, basically, you need to implement the onErrorNext and send your message to DLQ via StreamBridge and then return Mono.empty(), just bear in mind you need to use onErrorStop() to prevent any strange behaviour with your up/down stream errors so it will make sure the error goes to onErrorNext not to onErrorContinue. Here is the code snippet I used

@Bean
    @ConditionalOnProperty("app.reader.writerTest.enabled", havingValue = "true", matchIfMissing = true)
    fun writerTest(
        streamBridge: StreamBridge
    ) = Function<Flux<Message<TestReaderRequest>>, Mono<Void>> { item ->
        item
            .map {
                println("TEST: $it")
                it
            }
            .flatMap { request ->
                Mono
                    .just("Service Test")
                    .map {
                        throw Exception("Testing DLQ")
                    }
                    .handleDlq(request, streamBridge, logger, 1)
            }
            .then()
    }

fun <T> Mono<T>.handleDlq(request: Message<*>, streamBridge: StreamBridge, logger: Logger, retryThreshold: Int = 1) =
    onErrorResume { exception -> handleDlq<T>(request, retryThreshold, streamBridge, logger, exception) }
        .onErrorStop()

fun <T> Flux<T>.handleDlq(request: Message<*>, streamBridge: StreamBridge, logger: Logger, retryThreshold: Int = 1) =
    onErrorResume { exception -> handleDlq<T>(request, retryThreshold, streamBridge, logger, exception) }
        .onErrorStop()

private fun <T> handleDlq(
    request: Message<*>,
    retryThreshold: Int,
    streamBridge: StreamBridge,
    logger: Logger,
    exception: Throwable
): Mono<T> {
    logger.error(
        "Unable to finish the operation of request: $request due to: ",
        exception
    )
    val receivedTopic = request.headers[RECEIVED_TOPIC]
    val attempts = request.headers[DELIVERY_ATTEMPT, AtomicInteger::class.java] ?: AtomicInteger(1)
    val topic = if (attempts.get() < retryThreshold) {
        attempts.incrementAndGet()
        logger.info("About to retry the message of $request for topic: $receivedTopic, attempt #${attempts.get()} from $retryThreshold")
        receivedTopic.toString()
    } else {
        // TODO: needs to find the dlq name from application.yml
        val dlq = receivedTopic.toString() + "-dlq"
        logger.info("About to send the message of $request to dlq: $dlq after retrying ${attempts.get()} time(s)")
        dlq
    }
    val message = MessageBuilder
        .fromMessage(request)
        .setHeaders(buildRetryOrDlqMessageHeaders(request, attempts, topic, exception))
        .build()

    streamBridge.send(topic, message)
    return Mono.empty()
}

private fun buildRetryOrDlqMessageHeaders(
    request: Message<*>,
    attempts: AtomicInteger,
    topic: String,
    exception: Throwable
): MessageHeaderAccessor {
    val messageHeaders = MessageHeaderAccessor(request)

    messageHeaders.setHeader("spring.cloud.stream.sendto.destination", topic)
    messageHeaders.setHeader(DELIVERY_ATTEMPT, attempts)
    messageHeaders.setHeader(DLT_ORIGINAL_TOPIC, request.headers[RECEIVED_TOPIC])
    messageHeaders.setHeader(DLT_EXCEPTION_FQCN, exception.javaClass.name)
    messageHeaders.setHeader(DLT_EXCEPTION_MESSAGE, exception.message)
    messageHeaders.setHeader(DLT_EXCEPTION_STACKTRACE, exception.stackTraceToString())

    return messageHeaders
}

Let me know if you have any questions, hope it helps

Upvotes: 1

Related Questions