From aa94105f03b75eb2e0bf7d8b5f6bc601b07b9f25 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 18 Dec 2014 09:39:09 -0500 Subject: [PATCH] AMQP-456: Don't ERROR Log For Passive Declare JIRA: https://jira.spring.io/browse/AMQP-456 Previously all abnormal channel closes were logged at ERROR level, even NOT FOUND for passive declarations. --- .../AbstractCompressingPostProcessor.java | 11 ++- .../connection/CachingConnectionFactory.java | 7 +- .../amqp/rabbit/connection/RabbitUtils.java | 9 +++ .../core/RabbitTemplateIntegrationTests.java | 69 +++++++++++++++++++ 4 files changed, 94 insertions(+), 2 deletions(-) diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/AbstractCompressingPostProcessor.java b/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/AbstractCompressingPostProcessor.java index c62b7bb3..f94f8e82 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/AbstractCompressingPostProcessor.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/AbstractCompressingPostProcessor.java @@ -20,6 +20,9 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.OutputStream; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.amqp.AmqpException; import org.springframework.amqp.AmqpIOException; import org.springframework.amqp.core.Message; @@ -38,6 +41,8 @@ import org.springframework.util.FileCopyUtils; */ public abstract class AbstractCompressingPostProcessor implements MessagePostProcessor, Ordered { + private final Log logger = LogFactory.getLog(this.getClass()); + private final boolean autoDecompress; private int order; @@ -75,7 +80,11 @@ public abstract class AbstractCompressingPostProcessor implements MessagePostPro if (this.autoDecompress) { messageProperties.setHeader(MessageProperties.SPRING_AUTO_DECOMPRESS, true); } - return new Message(zipped.toByteArray(), messageProperties); + byte[] compressed = zipped.toByteArray(); + if (logger.isTraceEnabled()) { + logger.trace("Compressed " + message.getBody().length + " to " + compressed.length); + } + return new Message(compressed, messageProperties); } catch (IOException e) { throw new AmqpIOException(e); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java index 3ebaf33f..66970422 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java @@ -245,7 +245,12 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen @Override public void shutdownCompleted(ShutdownSignalException cause) { - if (!RabbitUtils.isNormalChannelClose(cause)) { + if (RabbitUtils.isPassiveDeclarationChannelClose(cause)) { + if (logger.isDebugEnabled()) { + logger.debug("Channel shutdown: " + cause.getMessage()); + } + } + else if (!RabbitUtils.isNormalChannelClose(cause)) { logger.error("Channel shutdown: " + cause.getMessage()); } } 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 5f444c86..faf92139 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 @@ -191,6 +191,15 @@ public abstract class RabbitUtils { && "OK".equals(((AMQP.Channel.Close) shutdownReason).getReplyText()); } + public static boolean isPassiveDeclarationChannelClose(ShutdownSignalException sig) { + Object shutdownReason = determineShutdownReason(sig); + return shutdownReason instanceof AMQP.Channel.Close + && AMQP.NOT_FOUND == ((AMQP.Channel.Close) shutdownReason).getReplyCode() + && ((((AMQP.Channel.Close) shutdownReason).getClassId() == 40 // exchange + || ((AMQP.Channel.Close) shutdownReason).getClassId() == 50) // queue + && ((AMQP.Channel.Close) shutdownReason).getMethodId() == 10); // declare + } + protected static Object determineShutdownReason(ShutdownSignalException sig) { if (shutDownSignalReasonMethod == null) { return false; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java index fab0bbf8..0b77049a 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java @@ -14,6 +14,7 @@ package org.springframework.amqp.rabbit.core; import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.startsWith; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -22,9 +23,14 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.mockito.Mockito.atLeast; import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.io.IOException; import java.io.UnsupportedEncodingException; import java.lang.reflect.Field; import java.util.HashMap; @@ -45,11 +51,13 @@ import org.junit.After; import org.junit.Before; import org.junit.Rule; import org.junit.Test; +import org.mockito.ArgumentCaptor; import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; import org.springframework.amqp.AmqpException; +import org.springframework.amqp.AmqpIOException; import org.springframework.amqp.core.Address; import org.springframework.amqp.core.AddressUtils; import org.springframework.amqp.core.Message; @@ -71,6 +79,8 @@ import org.springframework.amqp.support.converter.SimpleMessageConverter; import org.springframework.amqp.support.postprocessor.GUnzipPostProcessor; import org.springframework.amqp.support.postprocessor.GZipPostProcessor; import org.springframework.amqp.utils.SerializationUtils; +import org.springframework.amqp.utils.test.TestUtils; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.DisposableBean; import org.springframework.expression.common.LiteralExpression; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; @@ -88,6 +98,7 @@ import org.springframework.util.ReflectionUtils.FieldFilter; import com.rabbitmq.client.Channel; import com.rabbitmq.client.GetResponse; +import com.rabbitmq.client.ShutdownSignalException; /** * @author Dave Syer @@ -1140,6 +1151,64 @@ public class RabbitTemplateIntegrationTests { } } + @Test + public void testDegugLogOnPassiveDeclaration() { + CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost"); + Log logger = spy(TestUtils.getPropertyValue(connectionFactory, "logger", Log.class)); + when(logger.isDebugEnabled()).thenReturn(true); + new DirectFieldAccessor(connectionFactory).setPropertyValue("logger", logger); + RabbitTemplate template = new RabbitTemplate(connectionFactory); + final String queueName = UUID.randomUUID().toString(); + final String exchangeName = UUID.randomUUID().toString(); + try { + template.execute(new ChannelCallback() { + + @Override + public Void doInRabbit(Channel channel) throws Exception { + channel.queueDeclarePassive(queueName); + return null; + } + }); + fail("Expected exception"); + } + catch (Exception e) { + assertThat(e, instanceOf(AmqpIOException.class)); + assertThat(e.getCause(), instanceOf(IOException.class)); + assertThat(e.getCause().getCause(), instanceOf(ShutdownSignalException.class)); + assertThat(e.getCause().getCause().getMessage(), containsString("404")); + } + try { + template.execute(new ChannelCallback() { + + @Override + public Void doInRabbit(Channel channel) throws Exception { + channel.exchangeDeclarePassive(exchangeName); + return null; + } + }); + fail("Expected exception"); + } + catch (Exception e) { + assertThat(e, instanceOf(AmqpIOException.class)); + assertThat(e.getCause(), instanceOf(IOException.class)); + assertThat(e.getCause().getCause(), instanceOf(ShutdownSignalException.class)); + assertThat(e.getCause().getCause().getMessage(), containsString("404")); + } + verify(logger, never()).error(org.mockito.Matchers.any()); + ArgumentCaptor logs = ArgumentCaptor.forClass(Object.class); + verify(logger, atLeast(2)).debug(logs.capture()); + boolean queue = false; + boolean exchange = false; + for (Object log : logs.getAllValues()) { + String logMessage = (String) log; + queue |= (logMessage.contains(queueName) && logMessage.contains("404")); + exchange |= (logMessage.contains(queueName) && logMessage.contains("404")); + } + assertTrue(queue); + assertTrue(exchange); + connectionFactory.destroy(); + } + @SuppressWarnings("serial") private class PlannedException extends RuntimeException { public PlannedException() {