possibleFatalAuthenticationFailure flag - ignored by AsyncMessageProcessingConsumer

Question

Spring-AMQP added the possibleFailureAuthentication flag via AMQP-750. However, there is at least one case where the container can receive an AuthenticationFailureException and throw a FatalListenerStartupException, without checking that possibleFailureAuthentication flag (see lines 630-633 in BlockingQueueConsumer).

Is this an intended case? Whether or not it is, is there some way to end-run or catch and restart my container? In my environment, I'm making use of RabbitMQ's topology operator, so there can definitely be some timing deltas between Rabbit being up, services coming up, and users / vhosts / queues / exchanges all being set up. My next intended attempt is to test using the DirectMessageListenerContainer rather than the SimpleMessageListenerContainer.

Stack trace below, demonstrating use of spring-amqp 3.1.1 (via Spring Boot 3.2.2 and Spring Cloud Stream)

org.springframework.amqp.rabbit.listener.exception.FatalListenerStartupException: Authentication failure at org.springframework.amqp.rabbit.listener.BlockingQueueConsumer.start(BlockingQueueConsumer.java:632) ~[spring-rabbit-3.1.1.jar:3.1.1] at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.initialize(SimpleMessageListenerContainer.java:1436) ~[spring-rabbit-3.1.1.jar:3.1.1] at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1277) ~[spring-rabbit-3.1.1.jar:3.1.1] at java.base/java.lang.Thread.run(Unknown Source) ~[na:na] Caused by: org.springframework.amqp.AmqpAuthenticationException: com.rabbitmq.client.AuthenticationFailureException: ACCESS_REFUSED - Login was refused using authentication mechanism PLAIN. For details see the broker logfile. at org.springframework.amqp.rabbit.support.RabbitExceptionTranslator.convertRabbitAccessException(RabbitExceptionTranslator.java:64) ~[spring-rabbit-3.1.1.jar:3.1.1] at org.springframework.amqp.rabbit.connection.AbstractConnectionFactory.createBareConnection(AbstractConnectionFactory.java:606) ~[spring-rabbit-3.1.1.jar:3.1.1] at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.createConnection(CachingConnectionFactory.java:727) ~[spring-rabbit-3.1.1.jar:3.1.1] at org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils.createConnection(ConnectionFactoryUtils.java:257) ~[spring-rabbit-3.1.1.jar:3.1.1] at org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils$RabbitResourceFactory.createConnection(ConnectionFactoryUtils.java:345) ~[spring-rabbit-3.1.1.jar:3.1.1] at org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils.doGetTransactionalResourceHolder(ConnectionFactoryUtils.java:140) ~[spring-rabbit-3.1.1.jar:3.1.1] at org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils.getTransactionalResourceHolder(ConnectionFactoryUtils.java:102) ~[spring-rabbit-3.1.1.jar:3.1.1] at org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils.getTransactionalResourceHolder(ConnectionFactoryUtils.java:85) ~[spring-rabbit-3.1.1.jar:3.1.1] at org.springframework.amqp.rabbit.listener.BlockingQueueConsumer.start(BlockingQueueConsumer.java:626) ~[spring-rabbit-3.1.1.jar:3.1.1] ... 3 common frames omitted Caused by: com.rabbitmq.client.AuthenticationFailureException: ACCESS_REFUSED

• Login was refused using authentication mechanism PLAIN. For details see the broker logfile. at com.rabbitmq.client.impl.AMQConnection.start(AMQConnection.java:395) ~[amqp-client-5.19.0.jar:5.19.0] at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:1251) ~[amqp-client-5.19.0.jar:5.19.0] at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:1198) ~[amqp-client-5.19.0.jar:5.19.0] at org.springframework.amqp.rabbit.connection.AbstractConnectionFactory.connectAddresses(AbstractConnectionFactory.java:652) ~[spring-rabbit-3.1.1.jar:3.1.1] at org.springframework.amqp.rabbit.connection.AbstractConnectionFactory.connect(AbstractConnectionFactory.java:621) ~[spring-rabbit-3.1.1.jar:3.1.1] at org.springframework.amqp.rabbit.connection.AbstractConnectionFactory.createBareConnection(AbstractConnectionFactory.java:568) ~[spring-rabbit-3.1.1.jar:3.1.1] ... 10 common frames omitted

Answer 1

The BlockingQueueConsumer.start() is called from the AsyncMessageProcessingConsumer.initialize() where we have a logic like:

        catch (FatalListenerStartupException ex) {
            if (isPossibleAuthenticationFailureFatal()) {
                throw ex;
            }
            else {
                Throwable possibleAuthException = ex.getCause().getCause();
                if (!(possibleAuthException instanceof PossibleAuthenticationFailureException)) {
                    throw ex;
                }
                else {
                    this.start.countDown();
                    handleStartupFailure(this.consumer.getBackOffExecution());
                    throw possibleAuthException;
                }
            }
        }

According to your stack trace we really in that situation. Not sure what you think is wrong.