From 436ffcd75852d03b43a0d6f2c5c0ae8a1c03d492 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 12 May 2015 09:49:07 +0100 Subject: [PATCH] AMQP-498: Adjust Consumers With ChannelTransacted JIRA: https://jira.spring.io/browse/AMQP-498 In earlier versions, the while loop to drain the queue took an early exit when the channel was transacted. (This was actually incorrect since message delivery is not transactional, only acks). AMQP-388 introduced a new OR condition on the loop: `consumer.hasDelivery()`. This effectively made the `continuable` boolean in the OR condition irrelevant. This boolean was true if messages were received and the channel was not transactional. Another side effect was the `continuable` boolean was incorrectly used to adjust the consumers if the workload demanded. Instead, that decision should have depended just on whether messages were received; whether the channel is transacted is irrelevent. Remove the `continuable` boolean; fix the while loop and use a new boolean `receivedOk` which indicates whether messages were received and so whether we should consider increasing the consumers. Add a test to verify consumers are adjusted for transactional channels. --- .../listener/BlockingQueueConsumer.java | 3 +- .../SimpleMessageListenerContainer.java | 10 +-- ...mpleMessageListenerContainerLongTests.java | 61 ++++++++++++------- 3 files changed, 43 insertions(+), 31 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 55fb7ce1..56c4c867 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 @@ -786,7 +786,8 @@ public class BlockingQueueConsumer { RabbitUtils.commitIfNecessary(channel); } - } finally { + } + finally { deliveryTags.clear(); } 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 57233630..ebb1a499 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 @@ -1110,15 +1110,11 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta ConsumerChannelRegistry.registerConsumerChannel(consumer.getChannel(), getConnectionFactory()); } - // Always better to stop receiving as soon as possible if - // transactional - boolean continuable = false; - while (isActive(this.consumer) || this.consumer.hasDelivery() || continuable) { + while (isActive(this.consumer) || this.consumer.hasDelivery()) { try { - // Will come back false when the queue is drained - continuable = receiveAndExecute(this.consumer) && !isChannelTransacted(); + boolean receivedOk = receiveAndExecute(this.consumer); // At least one message received if (SimpleMessageListenerContainer.this.maxConcurrentConsumers != null) { - if (continuable) { + if (receivedOk) { if (isActive(this.consumer)) { consecutiveIdles = 0; if (consecutiveMessages++ > SimpleMessageListenerContainer.this.consecutiveActiveTrigger) { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerLongTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerLongTests.java index d88b820c..56bd1b4c 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerLongTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerLongTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2014 the original author or authors. + * Copyright 2013-2015 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. You may obtain a copy of the License at @@ -48,34 +48,49 @@ public class SimpleMessageListenerContainerLongTests { public LongRunningIntegrationTest longTest = new LongRunningIntegrationTest(); @Rule - public BrokerRunning brokerRunning = BrokerRunning.isRunning(); + public BrokerRunning brokerRunning = BrokerRunning.isRunningWithEmptyQueues("foo"); @Test public void testChangeConsumerCount() throws Exception { + testChangeConsumerCountGuts(false); + } + + @Test + public void testChangeConsumerCountTransacted() throws Exception { + testChangeConsumerCountGuts(true); + } + + private void testChangeConsumerCountGuts(boolean transacted) throws Exception { final SingleConnectionFactory singleConnectionFactory = new SingleConnectionFactory("localhost"); SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(singleConnectionFactory); - container.setMessageListener(new MessageListenerAdapter(this)); - container.setQueueNames("foo"); - container.setAutoStartup(false); - container.setConcurrentConsumers(2); - container.afterPropertiesSet(); - assertEquals(2, ReflectionTestUtils.getField(container, "concurrentConsumers")); - container.start(); - waitForNConsumers(container, 2); - container.setConcurrentConsumers(1); - waitForNConsumers(container, 1); - container.setMaxConcurrentConsumers(3); - RabbitTemplate template = new RabbitTemplate(singleConnectionFactory); - for (int i = 0; i < 20; i++) { - template.convertAndSend("foo", "foo"); + try { + container.setMessageListener(new MessageListenerAdapter(this)); + container.setQueueNames("foo"); + container.setAutoStartup(false); + container.setConcurrentConsumers(2); + container.setChannelTransacted(transacted); + container.afterPropertiesSet(); + assertEquals(2, ReflectionTestUtils.getField(container, "concurrentConsumers")); + container.start(); + waitForNConsumers(container, 2); + container.setConcurrentConsumers(1); + waitForNConsumers(container, 1); + container.setMaxConcurrentConsumers(3); + RabbitTemplate template = new RabbitTemplate(singleConnectionFactory); + for (int i = 0; i < 20; i++) { + template.convertAndSend("foo", "foo"); + } + waitForNConsumers(container, 2); // increased consumers due to work + waitForNConsumers(container, 1, 20000); // should stop the extra consumer after 10 seconds idle + container.setConcurrentConsumers(3); + waitForNConsumers(container, 3); + container.stop(); + waitForNConsumers(container, 0); + singleConnectionFactory.destroy(); + } + finally { + container.stop(); } - waitForNConsumers(container, 2); // increased consumers due to work - waitForNConsumers(container, 1, 20000); // should stop the extra consumer after 10 seconds idle - container.setConcurrentConsumers(3); - waitForNConsumers(container, 3); - container.stop(); - waitForNConsumers(container, 0); - singleConnectionFactory.destroy(); } @Test