GH-1347: DMLC Ignored Authentication Failures

Resolves #1347

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

**cherry-pick to 2.2.x**

# Conflicts:
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java
#	src/reference/asciidoc/amqp.adoc

Author: garyrussell

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

@@ -997,10 +997,14 @@ protected boolean isMismatchedQueuesFatal() {
 
 
 	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 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 DirectMessageListenerContainer() {
 	public DirectMessageListenerContainer(ConnectionFactory connectionFactory) {
 		setConnectionFactory(connectionFactory);
 		setMissingQueuesFatal(false);
+		doSetPossibleAuthenticationFailureFatal(false);
 	}
 
 	/**
@@ -418,6 +421,22 @@ protected void actualStart() {
 		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();
@@ -434,9 +453,7 @@ protected void actualStart() {
 		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,8 @@
 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;
 import static org.mockito.ArgumentMatchers.anyString;
@@ -45,6 +47,7 @@
 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 +129,15 @@ public void captureTestName(TestInfo info) {
 		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

@@ -5668,11 +5668,17 @@ a| image::images/tickmark.png[]
 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.