From 4cf9b5c2c5b08f4dbca16c7d94c4e7ac618f3a04 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 4 Oct 2023 12:33:24 -0400 Subject: [PATCH] GH-2536: Exclusive Consumer Logging Improvements Resolves https://github.com/spring-projects/spring-amqp/issues/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. --- .../support/ConditionalExceptionLogger.java | 14 +++++++++- .../connection/AbstractConnectionFactory.java | 11 +++----- .../amqp/rabbit/connection/RabbitUtils.java | 16 +++++++++++- .../AbstractMessageListenerContainer.java | 26 ++++--------------- .../DirectMessageListenerContainer.java | 18 ++++++------- .../SimpleMessageListenerContainer.java | 16 +++++++++--- ...ageListenerContainerIntegration2Tests.java | 21 ++++++++++----- src/reference/asciidoc/amqp.adoc | 17 ++++++++---- src/reference/asciidoc/whats-new.adoc | 12 +++++++-- 9 files changed, 93 insertions(+), 58 deletions(-) diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/ConditionalExceptionLogger.java b/spring-amqp/src/main/java/org/springframework/amqp/support/ConditionalExceptionLogger.java index f6845d7c..7d764fba 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/ConditionalExceptionLogger.java +++ b/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 @@ package org.springframework.amqp.support; 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); + } + } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java index 0735a691..312e7b20 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java @@ -496,7 +496,7 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di * 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 abstract class AbstractConnectionFactory implements ConnectionFactory, Di * 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 abstract class AbstractConnectionFactory implements ConnectionFactory, Di } } 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)) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java index 1b42b8c8..768f5e66 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java +++ b/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 abstract class RabbitUtils { } } + /** + * 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); + } + } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index 7392fdab..75a454dd 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java @@ -91,7 +91,6 @@ import org.springframework.util.backoff.BackOff; 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 abstract class AbstractMessageListenerContainer extends ObservableListene /** * 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 @@ public abstract class AbstractMessageListenerContainer extends ObservableListene * 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()); } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java index 5eb27853..ee3a02c7 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java @@ -782,24 +782,22 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta @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); + if (RabbitUtils.exclusiveAccesssRefused(ex)) { + getExclusiveConsumerExceptionLogger().log(logger, "Exclusive consumer failure", ex.getCause()); + publishConsumerFailedEvent("Consumer raised exception, attempting restart", false, ex); } - else if (e.getCause() instanceof ShutdownSignalException - && RabbitUtils.isPassiveDeclarationChannelClose((ShutdownSignalException) e.getCause())) { + 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) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java index 1eeec528..af48868e 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java +++ b/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.ListenerContainerAware; 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 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta private int consecutiveMessages; + private boolean failedExclusive; + AsyncMessageProcessingConsumer(BlockingQueueConsumer consumer) { this.consumer = consumer; @@ -1276,8 +1278,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } } 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 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } } 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); } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java index 0c247fc9..290b7d75 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java @@ -19,7 +19,6 @@ package org.springframework.amqp.rabbit.listener; 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.beans.factory.DisposableBean; 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 class SimpleMessageListenerContainerIntegration2Tests { @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 class SimpleMessageListenerContainerIntegration2Tests { consumeLatch1.countDown(); } }); + container1.setBeanName("container1"); container1.afterPropertiesSet(); container1.start(); assertThat(consumeLatch1.await(10, TimeUnit.SECONDS)).isTrue(); @@ -386,9 +387,10 @@ public class SimpleMessageListenerContainerIntegration2Tests { 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 @@ public class SimpleMessageListenerContainerIntegration2Tests { } assertThat(latch2.await(10, TimeUnit.SECONDS)).isTrue(); container2.stop(); - ArgumentCaptor captor = ArgumentCaptor.forClass(String.class); - verify(logger, atLeastOnce()).info(captor.capture()); - assertThat(captor.getAllValues()).anyMatch(arg -> arg.contains("exclusive")); + ArgumentCaptor 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 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 diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 202d47d7..879cf2a0 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -940,17 +940,19 @@ See <> 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 <>. [[runtime-cache-properties]] @@ -2364,8 +2366,13 @@ These events can be consumed by implementing `ApplicationListener>. Fatal errors are always logged at the `ERROR` level. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index cea3c5a0..ce30779d 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/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 <> and <> for more information.