rabbitmqspring-integrationspring-integration-amqp

How to handle errors when RabbitMQ exchange doesn't exist (and messages are sent through a messaging gateway interface)


I'd like to know what is the canonical way to handle errors in the following situation (code is a minimal working example):

@MessagingGateway(name = MY_GATEWAY, defaultRequestChannel = INPUT_CHANNEL)
public interface MyGateway
{
  @Gateway
  public void sendMessage(String message);
@Bean
public IntegrationFlow apiMutuaInputFlow()
{
  return IntegrationFlows
    .from(INPUT_CHANNEL)
    .handle(Amqp.outboundAdapter(rabbitConfig.myTemplate()))
    .get();
}
@Configuration
public class RabbitMqConfiguration
{
    @Autowired
    private ConnectionFactory rabbitConnectionFactory;

    @Bean
    public RabbitTemplate myTemplate()
    {
        RabbitTemplate r = new RabbitTemplate(rabbitConnectionFactory);
        r.setExchange(INPUT_QUEUE_NAME);
        r.setConnectionFactory(rabbitConnectionFactory);
        return r;
    }
}

I generally include a bean to define the RabbitMQ configuration I'm relying upon (exchange, queues and bindings), and it actually works fine. But while testing for failure scenarios, I found a situation I don't know how to properly handle using Spring Integration. The steps are:

What I would expect is:

What I find:

2020-02-11 08:18:40.746 ERROR 42778 --- [ 127.0.0.1:5672] o.s.a.r.c.CachingConnectionFactory       : Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'my.exchange' in vhost '/', class-id=60, method-id=40)

I understand such a failure may not be propagated upstream by the messaging gateway, whose job is isolating callers from the messaging API, but I definitely expect such an error to be interceptable.

Could you point me in the right direction?

Thank you.


Solution

  • RabbitMQ is inherently async, which is one reason that it performs so well.

    You can, however, block the caller by enabling confirms and returns and setting this option:

    /**
     * Set to true if you want to block the calling thread until a publisher confirm has
     * been received. Requires a template configured for returns. If a confirm is not
     * received within the confirm timeout or a negative acknowledgment or returned
     * message is received, an exception will be thrown. Does not apply to the gateway
     * since it blocks awaiting the reply.
     * @param waitForConfirm true to block until the confirmation or timeout is received.
     * @since 5.2
     * @see #setConfirmTimeout(long)
     * @see #setMultiSend(boolean)
     */
    public void setWaitForConfirm(boolean waitForConfirm) {
        this.waitForConfirm = waitForConfirm;
    }
    

    (With the DSL .waitForConfirm(true)).

    This also requires a confirm correlation expression. Here's an example from one of the test cases

        @Bean
        public IntegrationFlow flow(RabbitTemplate template) {
            return f -> f.handle(Amqp.outboundAdapter(template)
                    .exchangeName("")
                    .routingKeyFunction(msg -> msg.getHeaders().get("rk", String.class))
                    .confirmCorrelationFunction(msg -> msg)
                    .waitForConfirm(true));
        }
    
        @Bean
        public CachingConnectionFactory cf() {
            CachingConnectionFactory ccf = new CachingConnectionFactory(
                    RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
            ccf.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
            ccf.setPublisherReturns(true);
            return ccf;
        }
    
        @Bean
        public RabbitTemplate template(ConnectionFactory cf) {
            RabbitTemplate rabbitTemplate = new RabbitTemplate(cf);
            rabbitTemplate.setMandatory(true);               // for returns
            rabbitTemplate.setReceiveTimeout(10_000);
            return rabbitTemplate;
        }
    

    Bear in mind this will slow down things considerably (similar to using transactions) so you may want to reconsider whether you want to do this on every send (unless performance is not an issue).