GH-2536: Exclusive Consumer Logging Improvements

Resolves #2536

Log messages due to access refused due to exclusive consumers at DEBUG
level instead of WARN and INFO.

* Use LogMessage to avoid enabled check.

* Use LogMessage at INFO level too.

Author: garyrussell

spring-amqp/src/main/java/org/springframework/amqp/support/ConditionalExceptionLogger.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,6 +18,8 @@
 
 import org.apache.commons.logging.Log;
 
+import org.springframework.core.log.LogMessage;
+
 /**
  * For components that support customization of the logging of certain events, users can
  * provide an implementation of this interface to modify the existing logging behavior.
@@ -37,4 +39,14 @@ public interface ConditionalExceptionLogger {
 	 */
 	void log(Log logger, String message, Throwable t);
 
+	/**
+	 * Log a consumer restart; debug by default.
+	 * @param logger the logger.
+	 * @param message the message.
+	 * @since 3.1
+	 */
+	default void logRestart(Log logger, LogMessage message) {
+		logger.debug(message);
+	}
+
 }

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java

@@ -496,7 +496,7 @@ public void setConnectionNameStrategy(ConnectionNameStrategy connectionNameStrat
 	 * Set the strategy for logging close exceptions; by default, if a channel is closed due to a failed
 	 * passive queue declaration, it is logged at debug level. Normal channel closes (200 OK) are not
 	 * logged. All others are logged at ERROR level (unless access is refused due to an exclusive consumer
-	 * condition, in which case, it is logged at INFO level).
+	 * condition, in which case, it is logged at DEBUG level, since 3.1, previously INFO).
 	 * @param closeExceptionLogger the {@link ConditionalExceptionLogger}.
 	 * @since 1.5
 	 */
@@ -720,10 +720,7 @@ public void handleUnblocked() {
 	 * close exceptions.
 	 * @since 1.5
 	 */
-	private static class DefaultChannelCloseLogger implements ConditionalExceptionLogger {
-
-		DefaultChannelCloseLogger() {
-		}
+	public static class DefaultChannelCloseLogger implements ConditionalExceptionLogger {
 
 		@Override
 		public void log(Log logger, String message, Throwable t) {
@@ -734,8 +731,8 @@ public void log(Log logger, String message, Throwable t) {
 					}
 				}
 				else if (RabbitUtils.isExclusiveUseChannelClose(cause)) {
-					if (logger.isInfoEnabled()) {
-						logger.info(message + ": " + cause.getMessage());
+					if (logger.isDebugEnabled()) {
+						logger.debug(message + ": " + cause.getMessage());
 					}
 				}
 				else if (!RabbitUtils.isNormalChannelClose(cause)) {

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -409,4 +409,18 @@ public static SaslConfig stringToSaslConfig(String saslConfig,
 		}
 	}
 
+	/**
+	 * Determine whether the exception is due to an access refused for an exclusive consumer.
+	 * @param exception the exception.
+	 * @return true if access refused.
+	 * @since 3.1
+	 */
+	public static boolean exclusiveAccesssRefused(Exception exception) {
+		return exception.getCause() instanceof IOException
+				&& exception.getCause().getCause() instanceof ShutdownSignalException sse1
+				&& isExclusiveUseChannelClose(sse1)
+				|| exception.getCause() instanceof ShutdownSignalException sse2
+						&& isExclusiveUseChannelClose(sse2);
+	}
+
 }

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

@@ -91,7 +91,6 @@
 import org.springframework.util.backoff.FixedBackOff;
 
 import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.ShutdownSignalException;
 import io.micrometer.observation.Observation;
 import io.micrometer.observation.ObservationRegistry;
 
@@ -1030,7 +1029,7 @@ public void setStatefulRetryFatalWithNullMessageId(boolean statefulRetryFatalWit
 
 	/**
 	 * Set a {@link ConditionalExceptionLogger} for logging exclusive consumer failures. The
-	 * default is to log such failures at WARN level.
+	 * default is to log such failures at DEBUG level (since 3.1, previously WARN).
 	 * @param exclusiveConsumerExceptionLogger the conditional exception logger.
 	 * @since 1.5
 	 */
@@ -2095,27 +2094,12 @@ protected WrappedTransactionException(Throwable cause) {
 	 * consumer failures.
 	 * @since 1.5
 	 */
-	private static class DefaultExclusiveConsumerLogger implements ConditionalExceptionLogger {
-
-		DefaultExclusiveConsumerLogger() {
-		}
+	public static class DefaultExclusiveConsumerLogger implements ConditionalExceptionLogger {
 
 		@Override
-		public void log(Log logger, String message, Throwable t) {
-			if (t instanceof ShutdownSignalException cause) {
-				if (RabbitUtils.isExclusiveUseChannelClose(cause)) {
-					if (logger.isWarnEnabled()) {
-						logger.warn(message + ": " + cause.toString());
-					}
-				}
-				else if (!RabbitUtils.isNormalChannelClose(cause)) {
-					logger.error(message + ": " + cause.getMessage());
-				}
-			}
-			else {
-				if (logger.isErrorEnabled()) {
-					logger.error("Unexpected invocation of " + getClass() + ", with message: " + message, t);
-				}
+		public void log(Log logger, String message, Throwable cause) {
+			if (logger.isDebugEnabled()) {
+				logger.debug(message + ": " + cause.toString());
 			}
 		}
 

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

@@ -782,24 +782,22 @@ private SimpleConsumer consume(String queue, int index, Connection connection) {
 
 	@Nullable
 	private SimpleConsumer handleConsumeException(String queue, int index, @Nullable SimpleConsumer consumerArg,
-			Exception e) {
+			Exception ex) {
 
 		SimpleConsumer consumer = consumerArg;
-		if (e.getCause() instanceof ShutdownSignalException
-				&& e.getCause().getMessage().contains("in exclusive use")) {
-			getExclusiveConsumerExceptionLogger().log(logger,
-					"Exclusive consumer failure", e.getCause());
-			publishConsumerFailedEvent("Consumer raised exception, attempting restart", false, e);
-		}
-		else if (e.getCause() instanceof ShutdownSignalException
-				&& RabbitUtils.isPassiveDeclarationChannelClose((ShutdownSignalException) e.getCause())) {
+		if (RabbitUtils.exclusiveAccesssRefused(ex)) {
+			getExclusiveConsumerExceptionLogger().log(logger, "Exclusive consumer failure", ex.getCause());
+			publishConsumerFailedEvent("Consumer raised exception, attempting restart", false, ex);
+		}
+		else if (ex.getCause() instanceof ShutdownSignalException
+				&& RabbitUtils.isPassiveDeclarationChannelClose((ShutdownSignalException) ex.getCause())) {
 			publishMissingQueueEvent(queue);
 			this.logger.error("Queue not present, scheduling consumer "
-					+ (consumer == null ? "for queue " + queue : consumer) + " for restart", e);
+					+ (consumer == null ? "for queue " + queue : consumer) + " for restart", ex);
 		}
 		else if (this.logger.isWarnEnabled()) {
 			this.logger.warn("basicConsume failed, scheduling consumer "
-					+ (consumer == null ? "for queue " + queue : consumer) + " for restart", e);
+					+ (consumer == null ? "for queue " + queue : consumer) + " for restart", ex);
 		}
 
 		if (consumer == null) {

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

@@ -16,7 +16,6 @@
 
 package org.springframework.amqp.rabbit.listener;
 
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -57,6 +56,7 @@
 import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException;
 import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
 import org.springframework.amqp.support.ConsumerTagStrategy;
+import org.springframework.core.log.LogMessage;
 import org.springframework.jmx.export.annotation.ManagedMetric;
 import org.springframework.jmx.support.MetricType;
 import org.springframework.lang.Nullable;
@@ -1165,6 +1165,8 @@ private final class AsyncMessageProcessingConsumer implements Runnable {
 
 		private int consecutiveMessages;
 
+		private boolean failedExclusive;
+
 
 		AsyncMessageProcessingConsumer(BlockingQueueConsumer consumer) {
 			this.consumer = consumer;
@@ -1276,8 +1278,8 @@ public void run() { // NOSONAR - line count
 				}
 			}
 			catch (AmqpIOException e) {
-				if (e.getCause() instanceof IOException && e.getCause().getCause() instanceof ShutdownSignalException
-						&& e.getCause().getCause().getMessage().contains("in exclusive use")) {
+				if (RabbitUtils.exclusiveAccesssRefused(e)) {
+					this.failedExclusive = true;
 					getExclusiveConsumerExceptionLogger().log(logger,
 							"Exclusive consumer failure", e.getCause().getCause());
 					publishConsumerFailedEvent("Consumer raised exception, attempting restart", false, e);
@@ -1460,7 +1462,13 @@ private void killOrRestart(boolean aborted) {
 				}
 			}
 			else {
-				logger.info("Restarting " + this.consumer);
+				LogMessage restartMessage = LogMessage.of(() -> "Restarting " + this.consumer);
+				if (this.failedExclusive) {
+					getExclusiveConsumerExceptionLogger().logRestart(logger, restartMessage);
+				}
+				else {
+					logger.info(restartMessage);
+				}
 				restart(this.consumer);
 			}
 		}

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

@@ -19,7 +19,6 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 import static org.awaitility.Awaitility.with;
-import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willAnswer;
@@ -79,6 +78,7 @@
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.core.log.LogMessage;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 import com.rabbitmq.client.AMQP.Queue.DeclareOk;
@@ -347,7 +347,7 @@ public void testListenFromAnonQueue() throws Exception {
 	@Test
 	public void testExclusive() throws Exception {
 		Log logger = spy(TestUtils.getPropertyValue(this.template.getConnectionFactory(), "logger", Log.class));
-		willReturn(true).given(logger).isInfoEnabled();
+		willReturn(true).given(logger).isDebugEnabled();
 		new DirectFieldAccessor(this.template.getConnectionFactory()).setPropertyValue("logger", logger);
 		CountDownLatch latch1 = new CountDownLatch(1000);
 		SimpleMessageListenerContainer container1 =
@@ -365,6 +365,7 @@ public void testExclusive() throws Exception {
 				consumeLatch1.countDown();
 			}
 		});
+		container1.setBeanName("container1");
 		container1.afterPropertiesSet();
 		container1.start();
 		assertThat(consumeLatch1.await(10, TimeUnit.SECONDS)).isTrue();
@@ -386,9 +387,10 @@ else if (event instanceof ConsumeOkEvent) {
 				consumeLatch2.countDown();
 			}
 		});
+		container2.setBeanName("container2");
 		container2.afterPropertiesSet();
 		Log containerLogger = spy(TestUtils.getPropertyValue(container2, "logger", Log.class));
-		willReturn(true).given(containerLogger).isWarnEnabled();
+		willReturn(true).given(containerLogger).isDebugEnabled();
 		new DirectFieldAccessor(container2).setPropertyValue("logger", containerLogger);
 		container2.start();
 		for (int i = 0; i < 1000; i++) {
@@ -404,13 +406,18 @@ else if (event instanceof ConsumeOkEvent) {
 		}
 		assertThat(latch2.await(10, TimeUnit.SECONDS)).isTrue();
 		container2.stop();
-		ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
-		verify(logger, atLeastOnce()).info(captor.capture());
-		assertThat(captor.getAllValues()).anyMatch(arg -> arg.contains("exclusive"));
+		ArgumentCaptor<String> connLogCaptor = ArgumentCaptor.forClass(String.class);
+		verify(logger, atLeastOnce()).debug(connLogCaptor.capture());
+		assertThat(connLogCaptor.getAllValues()).anyMatch(arg -> arg.contains("exclusive"));
 		assertThat(eventRef.get().getReason()).isEqualTo("Consumer raised exception, attempting restart");
 		assertThat(eventRef.get().isFatal()).isFalse();
 		assertThat(eventRef.get().getThrowable()).isInstanceOf(AmqpIOException.class);
-		verify(containerLogger, atLeastOnce()).warn(any());
+		ArgumentCaptor<String> contLogCaptor = ArgumentCaptor.forClass(String.class);
+		verify(containerLogger, atLeastOnce()).debug(contLogCaptor.capture());
+		assertThat(contLogCaptor.getAllValues()).anyMatch(arg -> arg.contains("exclusive"));
+		ArgumentCaptor lmCaptor = ArgumentCaptor.forClass(LogMessage.class);
+		verify(containerLogger).debug(lmCaptor.capture());
+		assertThat(lmCaptor.getAllValues()).anyMatch(arg -> arg.toString().startsWith("Restarting "));
 	}
 
 	@Test

src/reference/asciidoc/amqp.adoc

@@ -940,17 +940,19 @@ See <<publishing-is-async>> for one scenario where you might want to register a
 
 Version 1.5 introduced a mechanism to enable users to control logging levels.
 
-The `CachingConnectionFactory` uses a default strategy to log channel closures as follows:
+The `AbstractConnectionFactory` uses a default strategy to log channel closures as follows:
 
 * Normal channel closes (200 OK) are not logged.
-* If a channel is closed due to a failed passive queue declaration, it is logged at debug level.
+* If a channel is closed due to a failed passive queue declaration, it is logged at DEBUG level.
 * If a channel is closed because the `basic.consume` is refused due to an exclusive consumer condition, it is logged at
-INFO level.
+DEBUG level (since 3.1, previously INFO).
 * All others are logged at ERROR level.
 
 To modify this behavior, you can inject a custom `ConditionalExceptionLogger` into the
 `CachingConnectionFactory` in its `closeExceptionLogger` property.
 
+Also, the `AbstractConnectionFactory.DefaultChannelCloseLogger` is now public, allowing it to be sub classed.
+
 See also <<consumer-events>>.
 
 [[runtime-cache-properties]]
@@ -2364,8 +2366,13 @@ These events can be consumed by implementing `ApplicationListener<ListenerContai
 
 NOTE: System-wide events (such as connection failures) are published by all consumers when `concurrentConsumers` is greater than 1.
 
-If a consumer fails because one if its queues is being used exclusively, by default, as well as publishing the event, a `WARN` log is issued.
-To change this logging behavior, provide a custom `ConditionalExceptionLogger` in the `SimpleMessageListenerContainer` instance's `exclusiveConsumerExceptionLogger` property.
+If a consumer fails because one if its queues is being used exclusively, by default, as well as publishing the event, a `DEBUG` log is issued (since 3.1, previously WARN).
+To change this logging behavior, provide a custom `ConditionalExceptionLogger` in the `AbstractMessageListenerContainer` instance's `exclusiveConsumerExceptionLogger` property.
+In addition, the `SimpleMessageListenerContainer` consumer restart after such an exception is now logged at DEBUG level by default (previously INFO).
+A new method `logRestart()` has been added to the `ConditionalExceptionLogger` to allow this to be changed.
+
+Also, the `AbstractMessageListenerContainer.DefaultExclusiveConsumerLogger` is now public, allowing it to be sub classed.
+
 See also <<channel-close-logging>>.
 
 Fatal errors are always logged at the `ERROR` level.

src/reference/asciidoc/whats-new.adoc

@@ -1,9 +1,17 @@
 [[whats-new]]
 == What's New
 
-=== Changes in 3.0 Since 2.4
+=== Changes in 3.1 Since 3.0
 
-==== Java 17, Spring Framework 6.0
+==== Java 17, Spring Framework 6.1
 
 This version requires Spring Framework 6.1 and Java 17.
 
+[[31-exc]]
+==== Exclusive Consumer Logging
+
+Log messages reporting access refusal due to exclusive consumers are now logged at DEBUG level by default.
+It remains possible to configure your own logging behavior by setting the `exclusiveConsumerExceptionLogger` and `closeExceptionLogger` properties on the listener container and connection factory respectively.
+In addition, the `SimpleMessageListenerContainer` consumer restart after such an exception is now logged at DEBUG level by default (previously INFO).
+A new method `logRestart()` has been added to the `ConditionalExceptionLogger` to allow this to be changed.
+See <<consumer-events>> and <<channel-close-logging>> for more information.