GH-1347: DMLC Ignored Authentication Failures

Resolves https://github.com/spring-projects/spring-amqp/issues/1347

Container starts and keeps trying to connect instead of honoring
the property.

**cherry-pick to 2.2.x**

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java

@@ -1001,10 +1001,14 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 
 
 	public void setPossibleAuthenticationFailureFatal(boolean possibleAuthenticationFailureFatal) {
-		this.possibleAuthenticationFailureFatal = possibleAuthenticationFailureFatal;
+		doSetPossibleAuthenticationFailureFatal(possibleAuthenticationFailureFatal);
 		this.possibleAuthenticationFailureFatalSet = true;
 	}
 
+	protected final void doSetPossibleAuthenticationFailureFatal(boolean possibleAuthenticationFailureFatal) {
+		this.possibleAuthenticationFailureFatal = possibleAuthenticationFailureFatal;
+	}
+
 	public boolean isPossibleAuthenticationFailureFatal() {
 		return this.possibleAuthenticationFailureFatal;
 	}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java

@@ -41,6 +41,7 @@ import java.util.stream.Stream;
 import org.apache.commons.logging.Log;
 
 import org.springframework.amqp.AmqpApplicationContextClosedException;
+import org.springframework.amqp.AmqpAuthenticationException;
 import org.springframework.amqp.AmqpConnectException;
 import org.springframework.amqp.AmqpException;
 import org.springframework.amqp.AmqpIOException;
@@ -145,6 +146,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 	 */
 	public DirectMessageListenerContainer() {
 		setMissingQueuesFatal(false);
+		doSetPossibleAuthenticationFailureFatal(false);
 	}
 
 	/**
@@ -154,6 +156,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 	public DirectMessageListenerContainer(ConnectionFactory connectionFactory) {
 		setConnectionFactory(connectionFactory);
 		setMissingQueuesFatal(false);
+		doSetPossibleAuthenticationFailureFatal(false);
 	}
 
 	/**
@@ -422,6 +425,22 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 		super.doStart();
 		final String[] queueNames = getQueueNames();
 		checkMissingQueues(queueNames);
+		if (isPossibleAuthenticationFailureFatal()) {
+			Connection connection = null;
+			try {
+				getConnectionFactory().createConnection();
+			}
+			catch (AmqpAuthenticationException ex) {
+				throw ex;
+			}
+			catch (Exception ex) { // NOSONAR
+			}
+			finally {
+				if (connection != null) {
+					connection.close();
+				}
+			}
+		}
 		long idleEventInterval = getIdleEventInterval();
 		if (this.taskScheduler == null) {
 			afterPropertiesSet();
@@ -438,9 +457,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 		if (queueNames.length > 0) {
 			doRedeclareElementsIfNecessary();
 			getTaskExecutor().execute(() -> { // NOSONAR never null here
-
 				startConsumers(queueNames);
-
 			});
 		}
 		else {

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java

@@ -17,6 +17,7 @@
 package org.springframework.amqp.rabbit.listener;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.awaitility.Awaitility.await;
 import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyMap;
@@ -45,6 +46,7 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInfo;
 import org.mockito.ArgumentCaptor;
 
+import org.springframework.amqp.AmqpAuthenticationException;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.Connection;
@@ -126,6 +128,15 @@ public class DirectMessageListenerContainerIntegrationTests {
 		this.testName = info.getDisplayName();
 	}
 
+	@Test
+	void authFailed() {
+		CachingConnectionFactory cf = new CachingConnectionFactory("localhost");
+		cf.setUsername("junk");
+		DirectMessageListenerContainer dmlc = new DirectMessageListenerContainer(cf);
+		dmlc.setPossibleAuthenticationFailureFatal(true);
+		assertThatExceptionOfType(AmqpAuthenticationException.class).isThrownBy(() -> dmlc.start());
+	}
+
 	@SuppressWarnings("unchecked")
 	@Test
 	public void testSimple() throws Exception {

src/reference/asciidoc/amqp.adoc

@@ -6146,11 +6146,17 @@ FailureFatal
 (possible-authentication-
 failure-fatal)
 
-a|When set to `true` (default), if a `PossibleAuthenticationFailureException` is thrown during connection, it is considered fatal.
-This causes the application context to fail to initialize during startup.
+a|When set to `true` (default for SMLC), if a `PossibleAuthenticationFailureException` is thrown during connection, it is considered fatal.
+This causes the application context to fail to initialize during startup (if the container is configured with auto startup).
 
 Since _version 2.0_.
 
+**DirectMessageListenerContainer**
+
+When set to `false` (default), each consumer will attempt to reconnect according to the `monitorInterval`.
+
+**SimpleMessageListenerContainer**
+
 When set to `false`, after making the 3 retries, the container will go into recovery mode, as with other problems, such as the broker being down.
 The container will attempt to recover according to the `recoveryInterval` property.
 During each recovery attempt, each consumer will again try 4 times to start.