Spring Cloud Stream Reactive - Handling Messages exception

Question

I'm currently working on a project in Kotlin that uses rabbit with reactor to receive messages of some DTO type, and dispatch them if they are up to certain criteria. In the process of testing my code, I tried to simulate bad Message input(Since the message is coming from an external service) and see the behaviour of the subscriber. Once I got a bad input message, the subscriber stopped listening to any new input and threw the following exception:

 org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application.calculateAverage-in-0'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage

I then tried to run the official example from spring and change the Supplier to send bad data on the first dispatch, then send valid data and see the behaviour.

On the Supplier's side, I have added an indexer to send a bad message only on the first run.

//Following source and sinks are used for testing only.
//Test source will send data to the same destination where the processor receives data
//Test sink will consume data from the same destination where the processor produces data
// ------ New Code -------
static int x = 0;
// ------ END New Code -------

static class TestSource {

    private AtomicBoolean semaphore = new AtomicBoolean(true);
    private Random random = new Random();
    private int[] ids = new int[]{100100, 100200, 100300};

    @Bean
    public Supplier sendTestData() {

        return () -> {
            // ------ New Code -------
            if(x==0) {
                return "hey";
            }
            x++;
            // ------ END New Code -------
            int id = ids[random.nextInt(3)];
            int temperature = random.nextInt((102 - 65) + 1) + 65;
            Sensor sensor = new Sensor();
            sensor.setId(id);
            sensor.setTemperature(temperature);
            return sensor;
        };
    }
}

The subscriber side:

@Bean
public Function, Flux> calculateAverage() {
    return data -> data.window(Duration.ofSeconds(3)).flatMap(
            window -> window.groupBy(Sensor::getId).flatMap(this::calculateAverage));
}

private Mono calculateAverage(GroupedFlux group) {
    return group
            .reduce(new Accumulator(0, 0),
                    (a, d) -> new Accumulator(a.getCount() + 1, a.getTotalValue() + d.getTemperature()))
            .map(accumulator -> new Average(group.key(), (accumulator.getTotalValue()) / accumulator.getCount()));
}

The results, as I suspected, showed that the subscriber fails to proceed with the next valid messages after failing on the bad input:

2021-02-21 17:46:57.905  INFO 30702 --- [lOCpiq_lPYTgA-1] o.s.c.s.m.DirectWithAttributesChannel    : Channel 'application.calculateAverage-in-0' has 0 subscriber(s).
2021-02-21 17:46:57.907 ERROR 30702 --- [lOCpiq_lPYTgA-1] onfiguration$FunctionToDestinationBinder : Failure was detected during execution of the reactive function 'calculateAverage'
2021-02-21 17:46:57.910 ERROR 30702 --- [lOCpiq_lPYTgA-1] reactor.core.publisher.Operators         : Operator called default onErrorDropped

org.springframework.messaging.converter.MessageConversionException: Could not read JSON: Unrecognized token 'hey': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')
 at [Source: (byte[])"hey"; line: 1, column: 4]; nested exception is com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'hey': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')
 at [Source: (byte[])"hey"; line: 1, column: 4]
    at org.springframework.messaging.converter.MappingJackson2MessageConverter.convertFromInternal(MappingJackson2MessageConverter.java:235) ~[spring-messaging-5.2.9.RELEASE.jar:5.2.9.RELEASE]
    at org.springframework.cloud.stream.converter.ApplicationJsonMessageMarshallingConverter.convertFromInternal(ApplicationJsonMessageMarshallingConverter.java:110) ~[spring-cloud-stream-3.0.11.RELEASE.jar:3.0.11.RELEASE]
    at org.springframework.messaging.converter.AbstractMessageConverter.fromMessage(AbstractMessageConverter.java:197) ~[spring-messaging-5.2.9.RELEASE.jar:5.2.9.RELEASE]
    at org.springframework.messaging.converter.CompositeMessageConverter.fromMessage(CompositeMessageConverter.java:70) ~[spring-messaging-5.2.9.RELEASE.jar:5.2.9.RELEASE]
    at org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry$FunctionInvocationWrapper.fromMessage(SimpleFunctionRegistry.java:932) ~[spring-cloud-function-context-3.0.13.RELEASE.jar:3.0.13.RELEASE]
    at org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry$FunctionInvocationWrapper.convertInputValueIfNecessary(SimpleFunctionRegistry.java:833) ~[spring-cloud-function-context-3.0.13.RELEASE.jar:3.0.13.RELEASE]
    at org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry$FunctionInvocationWrapper.lambda$convertInputPublisherIfNecessary$9(SimpleFunctionRegistry.java:772) ~[spring-cloud-function-context-3.0.13.RELEASE.jar:3.0.13.RELEASE]
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:100) ~[reactor-core-3.3.10.RELEASE.jar:3.3.10.RELEASE]
    at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:192) ~[reactor-core-3.3.10.RELEASE.jar:3.3.10.RELEASE]
    at reactor.core.publisher.EmitterProcessor.drain(EmitterProcessor.java:432) ~[reactor-core-3.3.10.RELEASE.jar:3.3.10.RELEASE]
    at reactor.core.publisher.EmitterProcessor.onNext(EmitterProcessor.java:274) ~[reactor-core-3.3.10.RELEASE.jar:3.3.10.RELEASE]
    at org.springframework.integration.util.IntegrationReactiveUtils.lambda$null$8(IntegrationReactiveUtils.java:133) ~[spring-integration-core-5.3.2.RELEASE.jar:5.3.2.RELEASE]
    at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:115) ~[spring-integration-core-5.3.2.RELEASE.jar:5.3.2.RELEASE]
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:133) ~[spring-integration-core-5.3.2.RELEASE.jar:5.3.2.RELEASE]
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:106) ~[spring-integration-core-5.3.2.RELEASE.jar:5.3.2.RELEASE]
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72) ~[spring-integration-core-5.3.2.RELEASE.jar:5.3.2.RELEASE]
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:570) ~[spring-integration-core-5.3.2.RELEASE.jar:5.3.2.RELEASE]
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:520) ~[spring-integration-core-5.3.2.RELEASE.jar:5.3.2.RELEASE]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187) ~[spring-messaging-5.2.9.RELEASE.jar:5.2.9.RELEASE]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166) ~[spring-messaging-5.2.9.RELEASE.jar:5.2.9.RELEASE]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47) ~[spring-messaging-5.2.9.RELEASE.jar:5.2.9.RELEASE]
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109) ~[spring-messaging-5.2.9.RELEASE.jar:5.2.9.RELEASE]
    at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:208) ~[spring-integration-core-5.3.2.RELEASE.jar:5.3.2.RELEASE]
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access$1300(AmqpInboundChannelAdapter.java:65) ~[spring-integration-amqp-5.3.2.RELEASE.jar:5.3.2.RELEASE]
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.lambda$onMessage$0(AmqpInboundChannelAdapter.java:294) ~[spring-integration-amqp-5.3.2.RELEASE.jar:5.3.2.RELEASE]
    at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287) ~[spring-retry-1.2.5.RELEASE.jar:na]
    at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:180) ~[spring-retry-1.2.5.RELEASE.jar:na]
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:290) ~[spring-integration-amqp-5.3.2.RELEASE.jar:5.3.2.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1591) ~[spring-rabbit-2.2.11.RELEASE.jar:2.2.11.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1510) ~[spring-rabbit-2.2.11.RELEASE.jar:2.2.11.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1498) ~[spring-rabbit-2.2.11.RELEASE.jar:2.2.11.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1489) ~[spring-rabbit-2.2.11.RELEASE.jar:2.2.11.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1433) ~[spring-rabbit-2.2.11.RELEASE.jar:2.2.11.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:975) ~[spring-rabbit-2.2.11.RELEASE.jar:2.2.11.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:921) ~[spring-rabbit-2.2.11.RELEASE.jar:2.2.11.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83) ~[spring-rabbit-2.2.11.RELEASE.jar:2.2.11.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1296) ~[spring-rabbit-2.2.11.RELEASE.jar:2.2.11.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1202) ~[spring-rabbit-2.2.11.RELEASE.jar:2.2.11.RELEASE]
    at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
2021-02-21 17:47:00.917 ERROR 30702 --- [lOCpiq_lPYTgA-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application.calculateAverage-in-0'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[3], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=sensor, amqp_deliveryTag=4, deliveryAttempt=3, amqp_consumerQueue=sensor.anonymous.Q3w6aZhbSlOCpiq_lPYTgA, amqp_redelivered=false, amqp_receivedRoutingKey=sensor, amqp_timestamp=Sun Feb 21 17:46:57 IST 2021, amqp_messageId=c6430a53-a916-524f-f436-9fa34f1ba4f9, id=6493e09a-4b4d-37f9-615d-24faa3f3b2f2, amqp_consumerTag=amq.ctag-LqdVMYDHHTDatSifcDm9bg, sourceData=(Body:'hey' MessageProperties [headers={}, timestamp=Sun Feb 21 17:46:57 IST 2021, messageId=c6430a53-a916-524f-f436-9fa34f1ba4f9, contentType=application/json, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=sensor, receivedRoutingKey=sensor, deliveryTag=4, consumerTag=amq.ctag-LqdVMYDHHTDatSifcDm9bg, consumerQueue=sensor.anonymous.Q3w6aZhbSlOCpiq_lPYTgA]), contentType=application/json, timestamp=1613922417903}], failedMessage=GenericMessage [payload=byte[3], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=sensor, amqp_deliveryTag=4, deliveryAttempt=3, amqp_consumerQueue=sensor.anonymous.Q3w6aZhbSlOCpiq_lPYTgA, amqp_redelivered=false, amqp_receivedRoutingKey=sensor, amqp_timestamp=Sun Feb 21 17:46:57 IST 2021, amqp_messageId=c6430a53-a916-524f-f436-9fa34f1ba4f9, id=6493e09a-4b4d-37f9-615d-24faa3f3b2f2, amqp_consumerTag=amq.ctag-LqdVMYDHHTDatSifcDm9bg, sourceData=(Body:'hey' MessageProperties [headers={}, timestamp=Sun Feb 21 17:46:57 IST 2021, messageId=c6430a53-a916-524f-f436-9fa34f1ba4f9, contentType=application/json, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=sensor, receivedRoutingKey=sensor, deliveryTag=4, consumerTag=amq.ctag-LqdVMYDHHTDatSifcDm9bg, consumerQueue=sensor.anonymous.Q3w6aZhbSlOCpiq_lPYTgA]), contentType=application/json, timestamp=1613922417903}]
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:76)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:570)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:520)
    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:65)
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.lambda$onMessage$0(AmqpInboundChannelAdapter.java:294)
    at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287)
    at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:180)
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:290)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1591)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1510)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1498)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1489)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1433)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:975)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:921)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1296)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1202)
    at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[3], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=sensor, amqp_deliveryTag=4, deliveryAttempt=3, amqp_consumerQueue=sensor.anonymous.Q3w6aZhbSlOCpiq_lPYTgA, amqp_redelivered=false, amqp_receivedRoutingKey=sensor, amqp_timestamp=Sun Feb 21 17:46:57 IST 2021, amqp_messageId=c6430a53-a916-524f-f436-9fa34f1ba4f9, id=6493e09a-4b4d-37f9-615d-24faa3f3b2f2, amqp_consumerTag=amq.ctag-LqdVMYDHHTDatSifcDm9bg, sourceData=(Body:'hey' MessageProperties [headers={}, timestamp=Sun Feb 21 17:46:57 IST 2021, messageId=c6430a53-a916-524f-f436-9fa34f1ba4f9, contentType=application/json, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=sensor, receivedRoutingKey=sensor, deliveryTag=4, consumerTag=amq.ctag-LqdVMYDHHTDatSifcDm9bg, consumerQueue=sensor.anonymous.Q3w6aZhbSlOCpiq_lPYTgA]), contentType=application/json, timestamp=1613922417903}]
    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

2021-02-21 17:47:03.507  INFO 30702 --- [extShutdownHook] o.s.a.r.l.SimpleMessageListenerContainer : Waiting for workers to finish.
2021-02-21 17:47:03.936 ERROR 30702 --- [lOCpiq_lPYTgA-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application.calculateAverage-in-0'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[3], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=sensor, amqp_deliveryTag=5, deliveryAttempt=3, amqp_consumerQueue=sensor.anonymous.Q3w6aZhbSlOCpiq_lPYTgA, amqp_redelivered=false, amqp_receivedRoutingKey=sensor, amqp_timestamp=Sun Feb 21 17:46:58 IST 2021, amqp_messageId=b67652f5-1842-4c13-596d-295b36002217, id=4f2a90f9-5aae-0fae-507d-787694f605dc, amqp_consumerTag=amq.ctag-LqdVMYDHHTDatSifcDm9bg, sourceData=(Body:'hey' MessageProperties [headers={}, timestamp=Sun Feb 21 17:46:58 IST 2021, messageId=b67652f5-1842-4c13-596d-295b36002217, contentType=application/json, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=sensor, receivedRoutingKey=sensor, deliveryTag=5, consumerTag=amq.ctag-LqdVMYDHHTDatSifcDm9bg, consumerQueue=sensor.anonymous.Q3w6aZhbSlOCpiq_lPYTgA]), contentType=application/json, timestamp=1613922420930}], failedMessage=GenericMessage [payload=byte[3], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=sensor, amqp_deliveryTag=5, deliveryAttempt=3, amqp_consumerQueue=sensor.anonymous.Q3w6aZhbSlOCpiq_lPYTgA, amqp_redelivered=false, amqp_receivedRoutingKey=sensor, amqp_timestamp=Sun Feb 21 17:46:58 IST 2021, amqp_messageId=b67652f5-1842-4c13-596d-295b36002217, id=4f2a90f9-5aae-0fae-507d-787694f605dc, amqp_consumerTag=amq.ctag-LqdVMYDHHTDatSifcDm9bg, sourceData=(Body:'hey' MessageProperties [headers={}, timestamp=Sun Feb 21 17:46:58 IST 2021, messageId=b67652f5-1842-4c13-596d-295b36002217, contentType=application/json, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=sensor, receivedRoutingKey=sensor, deliveryTag=5, consumerTag=amq.ctag-LqdVMYDHHTDatSifcDm9bg, consumerQueue=sensor.anonymous.Q3w6aZhbSlOCpiq_lPYTgA]), contentType=application/json, timestamp=1613922420930}]
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:76)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:570)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:520)
    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:65)
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.lambda$onMessage$0(AmqpInboundChannelAdapter.java:294)
    at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287)
    at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:180)
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:290)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1591)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1510)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1498)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1489)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1433)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:975)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:921)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1296)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1202)
    at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[3], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=sensor, amqp_deliveryTag=5, deliveryAttempt=3, amqp_consumerQueue=sensor.anonymous.Q3w6aZhbSlOCpiq_lPYTgA, amqp_redelivered=false, amqp_receivedRoutingKey=sensor, amqp_timestamp=Sun Feb 21 17:46:58 IST 2021, amqp_messageId=b67652f5-1842-4c13-596d-295b36002217, id=4f2a90f9-5aae-0fae-507d-787694f605dc, amqp_consumerTag=amq.ctag-LqdVMYDHHTDatSifcDm9bg, sourceData=(Body:'hey' MessageProperties [headers={}, timestamp=Sun Feb 21 17:46:58 IST 2021, messageId=b67652f5-1842-4c13-596d-295b36002217, contentType=application/json, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=sensor, receivedRoutingKey=sensor, deliveryTag=5, consumerTag=amq.ctag-LqdVMYDHHTDatSifcDm9bg, consumerQueue=sensor.anonymous.Q3w6aZhbSlOCpiq_lPYTgA]), contentType=application/json, timestamp=1613922420930}]
    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

My question is how do I handle exceptions that are happening 'behind the scenes' such as the input parsing?

user_837717 · Answer

This is a tough question, given the nature of reactive programming, so we may need to take this discussion into an issue, so feel free to raise one, but here is my take.

The fundamental difference between reactive and imperative function is the concept of unit-of-work. With imperative function the unit-of-work is a single Message, so the framework maintains constant control of the stream only passing it for an instant via Message to a function. So you would expect and rightfully so that we would have something for error handling regardless where error happened - and we do.

With reactive function the world changes completely since the unit-of-work is the entire stream and function serves only as a connector between stream provided by the framework and stream operations defined by the user. At this point, s-c-stream has no control over what user does, so our general recommendation especially given the richness of reactive API when it comes to error handling is for the user to handle it on their own. But understand it is not because we don't want to, rather we can't since we have no view into the stream at that point. Your issue is indeed rather unique, since exception happens before steps defined by you are executed, specifically the type conversion we provide. Indeed there are things we can do to assist with that, but we're still searching for a consensus as to what those things should be and until we do fail-fast is the solution. You can overcome it by fixing your input since it is obviously not JSON and/or relax your function signature Function, Flux> and take care of type conversion yourself.

Anyway, as you can see I am open to suggestions, so feel free to raise and issue and provide your input.

user_10724097 · Answer

@Bean
Function, Flux> myFunction() {
    return messageFlux
        .onErrorContinue(MessageConversionException.class, (e, obj) -> {log.warn(...)})
        ...
}

This is what worked for me before. It will allow the upstream to continue on a conversion exception.

user_329589 · Answer

Here's my take and the strategies used to handle all this:

First I try to redirect known bad inputs to an error topic (like it was already suggested here):

private Flux knownErrorsHandling(Flux pointFlux) {
        return pointFlux.onErrorContinue(
            this::isKnownExceptionThatShouldBeRedirectedToErrorsChannel,
            (e, o) -> streamBridge.send(appProperties.getErrorsBindingName(), o));
    }

When an exception is not handled the subscription is cancelled. That's why we then see the Dispatcher has no subscribers error.

A weird side effect of not re-subscribing after loosing the subscription is that after a couple of tries to process the message and after throwing the Dispatcher has no subscribers exception the Kafka message is acknowledged, effectively loosing the message.

We could make use of the Flux#retry() variants, but that only makes the subscriber re-subscribe, without trying to re-process the same message.

My approach was to avoid Reactor retries and instead exit the application if an unrecoverable exception occurs, relying on the underlying platform to restart the application in the hope of it working after the recycling:

    @EventListener
    public void onApplicationStartedRegisterReactorHookToStopApplicationIfErrorDropped(ApplicationStartedEvent e) {
        Hooks.onErrorDropped(t -> {
            LOG.error("Unable to recover from exception.", t);
            e.getApplicationContext().close();
            System.exit(1);
        });
    }

I also have the auto commit offsets turned off: spring.cloud.stream.kafka.bindings..consumer.autoCommitOffset: false, especially as I rely on the Reactor's .bufferTimeout, which I noticed triggers message acknowledgment for each message, even though the batch might fail.

Overall I feel as the reactive approach is not the way to go with Spring Cloud Streams. At least not for now.

Spring Cloud Stream Reactive - Handling Messages exception

3 Answers