Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2536: Exclusive Consumer Logging Improvements

Browse Source 
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.
This commit is contained in:
Gary Russell
2023-10-04 12:33:24 -04:00
committed by GitHub
parent f6d46f69b3
commit 4cf9b5c2c5
9 changed files with 93 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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);
}
}

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

@@ -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)) {

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

@@ -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);
}
}

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

@@ -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());
}
}

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

@@ -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) {

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

@@ -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);
}
}

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

@@ -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<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

17
src/reference/asciidoc/amqp.adoc
View File

@@ -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.

12
src/reference/asciidoc/whats-new.adoc
View File

@@ -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.