From e535c33cf210e7b5d0d02e8d30ab44207fb56970 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 19 Mar 2019 13:39:49 -0400 Subject: [PATCH] GH-935: Handle all exceptions in handleDelivery Fixes https://github.com/spring-projects/spring-amqp/issues/935 - Don't call basicCancel if already canceled - Catch all `Exception`s # Conflicts: # spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java # spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java --- .../listener/BlockingQueueConsumer.java | 21 ++++++++++------ .../listener/BlockingQueueConsumerTests.java | 25 +++++++++++++------ 2 files changed, 31 insertions(+), 15 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java index a521df19..296280ed 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java @@ -843,7 +843,9 @@ public class BlockingQueueConsumer { private final class InternalConsumer extends DefaultConsumer { - private InternalConsumer(Channel channel) { + boolean canceled; + + InternalConsumer(Channel channel) { super(channel); } @@ -890,11 +892,12 @@ public class BlockingQueueConsumer { + BlockingQueueConsumer.this.consumerTags.get(consumerTag) + "); " + BlockingQueueConsumer.this); } + this.canceled = true; } @Override - public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) - throws IOException { + public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, + byte[] body) { if (logger.isDebugEnabled()) { logger.debug("Storing delivery for consumerTag: '" + consumerTag + "' with deliveryTag: '" + envelope.getDeliveryTag() + "' in " @@ -907,12 +910,13 @@ public class BlockingQueueConsumer { RabbitUtils.setPhysicalCloseRequired(getChannel(), true); // Defensive - should never happen BlockingQueueConsumer.this.queue.clear(); - getChannel().basicNack(envelope.getDeliveryTag(), true, true); - getChannel().basicCancel(consumerTag); + if (!this.canceled) { + getChannel().basicCancel(consumerTag); + } try { getChannel().close(); } - catch (TimeoutException e) { + catch (@SuppressWarnings("unused") TimeoutException e) { // no-op } } @@ -921,9 +925,12 @@ public class BlockingQueueConsumer { BlockingQueueConsumer.this.queue.put(new Delivery(consumerTag, envelope, properties, body)); } } - catch (InterruptedException e) { + catch (@SuppressWarnings("unused") InterruptedException e) { Thread.currentThread().interrupt(); } + catch (Exception e) { + BlockingQueueConsumer.logger.warn("Unexpected exception during delivery", e); + } } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java index 96685aaf..73b10a89 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java @@ -26,6 +26,7 @@ import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyBoolean; import static org.mockito.Matchers.anyString; import static org.mockito.Matchers.eq; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; @@ -40,8 +41,10 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.apache.log4j.Level; import org.junit.Rule; @@ -316,8 +319,7 @@ public class BlockingQueueConsumerTests { } @Test - @SuppressWarnings("unchecked") - public void testDrainAndReject() throws IOException { + public void testDrainAndReject() throws IOException, TimeoutException { ConnectionFactory connectionFactory = mock(ConnectionFactory.class); Connection connection = mock(Connection.class); ChannelProxy channel = mock(ChannelProxy.class); @@ -330,9 +332,18 @@ public class BlockingQueueConsumerTests { doReturn(isOpen.get()).when(channel).isOpen(); when(channel.queueDeclarePassive(anyString())) .then(invocation -> mock(AMQP.Queue.DeclareOk.class)); - when(channel.basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(), - any(Map.class), any(Consumer.class))).thenReturn("consumerTag"); - + AtomicReference theConsumer = new AtomicReference<>(); + doAnswer(inv -> { + Consumer consumer = (Consumer) inv.getArguments()[6]; + consumer.handleConsumeOk("consumerTag"); + theConsumer.set(consumer); + return "consumerTag"; + }).when(channel).basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(), + any(), any(Consumer.class)); + doAnswer(inv -> { + theConsumer.get().handleCancelOk("consumerTag"); + return null; + }).when(channel).basicCancel("consumerTag"); BlockingQueueConsumer blockingQueueConsumer = new BlockingQueueConsumer(connectionFactory, new DefaultMessagePropertiesConverter(), new ActiveObjectCounter(), AcknowledgeMode.AUTO, true, 2, "test"); @@ -357,9 +368,7 @@ public class BlockingQueueConsumerTests { envelope = new Envelope(3, false, "foo", "bar"); consumer.handleDelivery("consumerTag", envelope, props, new byte[0]); assertThat(TestUtils.getPropertyValue(blockingQueueConsumer, "queue", BlockingQueue.class).size(), equalTo(0)); - verify(channel).basicNack(3, true, true); - verify(channel, times(2)).basicCancel("consumerTag"); + verify(channel, times(1)).basicCancel("consumerTag"); } - }