From ba6875de9920b498c99ce43732bf01871f48241e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 28 Jun 2019 10:17:22 -0400 Subject: [PATCH] GH-1034: DMLC: Cancel consumer after failed ack Resolves https://github.com/spring-projects/spring-amqp/issues/1034 The monitor task now cancels the consumer after a failed ack/nack, whether or not the channel `isOpen()` returns true. Test with a mock channel that stays open after a failed ack. **cherry-pick to 2.1.x, 2.0.x** * Fix tests removing AssertJ dependency --- .../DirectMessageListenerContainer.java | 27 +++++++++-- ...rectMessageListenerContainerMockTests.java | 46 +++++++++++++++++++ 2 files changed, 69 insertions(+), 4 deletions(-) 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 99d0fc26..b133cfdb 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 @@ -387,7 +387,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta synchronized (this.consumersMonitor) { consumersToCancel = this.consumers.stream() .filter(c -> { - boolean open = c.getChannel().isOpen(); + boolean open = c.getChannel().isOpen() && !c.isAckFailed(); if (open && this.messagesPerAck > 1) { try { c.ackIfNecessary(now); @@ -502,7 +502,6 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta } } } - }); } else { @@ -717,7 +716,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta this.logger.debug("Canceling " + consumer); } synchronized (consumer) { - consumer.canceled = true; + consumer.setCanceled(true); if (this.messagesPerAck > 1) { consumer.ackIfNecessary(0L); } @@ -791,6 +790,8 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta private volatile boolean canceled; + private volatile boolean ackFailed; + private SimpleConsumer(Connection connection, Channel channel, String queue) { super(channel); this.connection = connection; @@ -815,6 +816,23 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta return this.epoch; } + /** + * Set to true to indicate this consumer is canceled and should send any pending + * acks. + * @param canceled the canceled to set + */ + void setCanceled(boolean canceled) { + this.canceled = canceled; + } + + /** + * True if an ack/nack failed (probably due to a closed channel). + * @return the ackFailed + */ + boolean isAckFailed() { + return this.ackFailed; + } + /** * Increment and return the current epoch for this consumer; consumersMonitor must * be held. @@ -966,6 +984,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta } } catch (Exception e) { + this.ackFailed = true; this.logger.error("Error acking", e); } } @@ -976,7 +995,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta * @param now the current time. * @throws IOException if one occurs. */ - private synchronized void ackIfNecessary(long now) throws IOException { + synchronized void ackIfNecessary(long now) throws IOException { if (this.pendingAcks >= this.messagesPerAck || ( this.pendingAcks > 0 && (now - this.lastAck > this.ackTimeout || this.canceled))) { sendAck(now); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerMockTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerMockTests.java index b8a00e7b..f5107f39 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerMockTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerMockTests.java @@ -28,6 +28,7 @@ import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.BDDMockito.willThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -42,6 +43,7 @@ import java.util.concurrent.atomic.AtomicReference; import org.junit.Test; import org.mockito.Mockito; +import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.rabbit.connection.ChannelProxy; import org.springframework.amqp.rabbit.connection.Connection; import org.springframework.amqp.rabbit.connection.ConnectionFactory; @@ -265,6 +267,50 @@ public class DirectMessageListenerContainerMockTests { container.stop(); } + @Test + public void testMonitorCancelsAfterBadAckEvenIfChannelReportsOpen() throws Exception { + ConnectionFactory connectionFactory = mock(ConnectionFactory.class); + Connection connection = mock(Connection.class); + ChannelProxy channel = mock(ChannelProxy.class); + Channel rabbitChannel = mock(Channel.class); + given(channel.getTargetChannel()).willReturn(rabbitChannel); + + given(connectionFactory.createConnection()).willReturn(connection); + given(connection.createChannel(anyBoolean())).willReturn(channel); + given(channel.isOpen()).willReturn(true); + given(channel.queueDeclarePassive(Mockito.anyString())) + .willAnswer(invocation -> mock(AMQP.Queue.DeclareOk.class)); + AtomicReference consumer = new AtomicReference<>(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + willAnswer(inv -> { + consumer.set(inv.getArgument(6)); + latch1.countDown(); + return "consumerTag"; + }).given(channel).basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(), + anyMap(), any(Consumer.class)); + + willThrow(new RuntimeException("bad ack")).given(channel).basicAck(1L, false); + willAnswer(inv -> { + consumer.get().handleCancelOk("consumerTag"); + latch2.countDown(); + return null; + }).given(channel).basicCancel("consumerTag"); + + DirectMessageListenerContainer container = new DirectMessageListenerContainer(connectionFactory); + container.setQueueNames("test"); + container.setPrefetchCount(2); + container.setMonitorInterval(100); + container.setMessageListener(mock(MessageListener.class)); + container.afterPropertiesSet(); + container.start(); + + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + consumer.get().handleDelivery("consumerTag", envelope(1L), new BasicProperties(), new byte[1]); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + container.stop(); + } + private Envelope envelope(long tag) { return new Envelope(tag, false, "", ""); }