GH-1318: Support the Global Flag in basicQos

garyrussell · artembilan · commit dcaa6967c33a · 2021-04-07T17:47:25.000-04:00
Resolves #1318 **cherry-pick to 2.2.x** # Conflicts: # spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java # spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java # src/reference/asciidoc/amqp.adoc # src/reference/asciidoc/whats-new.adoc
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2020 the original author or authors.
+ * Copyright 2014-2021 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.
@@ -87,6 +87,8 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 
 	private Integer prefetchCount;
 
+	private Boolean globalQos;
+
 	private Boolean defaultRequeueRejected;
 
 	private Advice[] adviceChain;
@@ -400,6 +402,16 @@ public void setDeBatchingEnabled(final Boolean deBatchingEnabled) {
 		this.deBatchingEnabled = deBatchingEnabled;
 	}
 
+	/**
+	 * Apply prefetch to the entire channel.
+	 * @param globalQos true for a channel-wide prefetch.
+	 * @since 2.2.17
+	 * @see com.rabbitmq.client.Channel#basicQos(int, boolean)
+	 */
+	public void setGlobalQos(boolean globalQos) {
+		this.globalQos = globalQos;
+	}
+
 	@Override
 	public C createListenerContainer(RabbitListenerEndpoint endpoint) {
 		C instance = createContainerInstance();
@@ -418,6 +430,7 @@ public C createListenerContainer(RabbitListenerEndpoint endpoint) {
 			.acceptIfNotNull(this.taskExecutor, instance::setTaskExecutor)
 			.acceptIfNotNull(this.transactionManager, instance::setTransactionManager)
 			.acceptIfNotNull(this.prefetchCount, instance::setPrefetchCount)
+			.acceptIfNotNull(this.globalQos, instance::setGlobalQos)
 			.acceptIfNotNull(this.defaultRequeueRejected, instance::setDefaultRequeueRejected)
 			.acceptIfNotNull(this.adviceChain, instance::setAdviceChain)
 			.acceptIfNotNull(this.recoveryBackOff, instance::setRecoveryBackOff)
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -48,6 +48,8 @@
 import org.springframework.util.ErrorHandler;
 import org.springframework.util.backoff.BackOff;
 
+import com.rabbitmq.client.Channel;
+
 /**
  * A Factory bean to create a listener container.
  *
@@ -111,6 +113,8 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean<AbstractMe
 
 	private Integer prefetchCount;
 
+	private Boolean globalQos;
+
 	private Long shutdownTimeout;
 
 	private Long idleEventInterval;
@@ -268,6 +272,16 @@ public void setPrefetchCount(int prefetchCount) {
 		this.prefetchCount = prefetchCount;
 	}
 
+	/**
+	 * Apply prefetch to the entire channel.
+	 * @param globalQos true for a channel-wide prefetch.
+	 * @since 2.2.17
+	 * @see Channel#basicQos(int, boolean)
+	 */
+	public void setGlobalQos(boolean globalQos) {
+		this.globalQos = globalQos;
+	}
+
 	public void setShutdownTimeout(long shutdownTimeout) {
 		this.shutdownTimeout = shutdownTimeout;
 	}
@@ -460,6 +474,7 @@ protected AbstractMessageListenerContainer createInstance() { // NOSONAR complex
 					.acceptIfNotNull(this.exclusive, container::setExclusive)
 					.acceptIfNotNull(this.defaultRequeueRejected, container::setDefaultRequeueRejected)
 					.acceptIfNotNull(this.prefetchCount, container::setPrefetchCount)
+					.acceptIfNotNull(this.globalQos, container::setGlobalQos)
 					.acceptIfNotNull(this.shutdownTimeout, container::setShutdownTimeout)
 					.acceptIfNotNull(this.idleEventInterval, container::setIdleEventInterval)
 					.acceptIfNotNull(this.transactionManager, container::setTransactionManager)
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceUtils.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -68,6 +68,8 @@ public final class RabbitNamespaceUtils {
 
 	private static final String PREFETCH_ATTRIBUTE = "prefetch";
 
+	private static final String GLOBAL_QOS = "global-qos";
+
 	private static final String RECEIVE_TIMEOUT_ATTRIBUTE = "receive-timeout";
 
 	private static final String CHANNEL_TRANSACTED_ATTRIBUTE = "channel-transacted";
@@ -204,6 +206,11 @@ public static BeanDefinition parseContainer(Element containerEle, ParserContext
 			containerDef.getPropertyValues().add("prefetchCount", new TypedStringValue(prefetch));
 		}
 
+		String globalQos = containerEle.getAttribute(GLOBAL_QOS);
+		if (StringUtils.hasText(globalQos)) {
+			containerDef.getPropertyValues().add("globalQos", new TypedStringValue(globalQos));
+		}
+
 		String receiveTimeout = containerEle.getAttribute(RECEIVE_TIMEOUT_ATTRIBUTE);
 		if (StringUtils.hasText(receiveTimeout)) {
 			containerDef.getPropertyValues().add("receiveTimeout", new TypedStringValue(receiveTimeout));
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java
@@ -220,6 +220,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 
 	private volatile int prefetchCount = DEFAULT_PREFETCH_COUNT;
 
+	private boolean globalQos;
+
 	private long idleEventInterval;
 
 	private volatile long lastReceive = System.currentTimeMillis();
@@ -793,6 +795,7 @@ protected boolean isDefaultRequeueRejected() {
 	 * Tell the broker how many messages to send to each consumer in a single request.
 	 * Often this can be set quite high to improve throughput.
 	 * @param prefetchCount the prefetch count
+	 * @see com.rabbitmq.client.Channel#basicQos(int, boolean)
 	 */
 	public void setPrefetchCount(int prefetchCount) {
 		this.prefetchCount = prefetchCount;
@@ -807,6 +810,20 @@ protected int getPrefetchCount() {
 		return this.prefetchCount;
 	}
 
+	/**
+	 * Apply prefetchCount to the entire channel.
+	 * @param globalQos true for a channel-wide prefetch.
+	 * @since 2.2.17
+	 * @see com.rabbitmq.client.Channel#basicQos(int, boolean)
+	 */
+	public void setGlobalQos(boolean globalQos) {
+		this.globalQos = globalQos;
+	}
+
+	protected boolean isGlobalQos() {
+		return this.globalQos;
+	}
+
 	/**
 	 * The time to wait for workers in milliseconds after the container is stopped. If any
 	 * workers are active when the shutdown signal comes they will be allowed to finish
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
@@ -160,6 +160,8 @@ public class BlockingQueueConsumer {
 
 	private java.util.function.Consumer<String> missingQueuePublisher = str -> { };
 
+	private boolean globalQos;
+
 	private volatile long abortStarted;
 
 	private volatile boolean normalCancel;
@@ -392,6 +394,16 @@ public void clearDeliveryTags() {
 		this.deliveryTags.clear();
 	}
 
+	/**
+	 * Apply prefetch to the entire channel.
+	 * @param globalQos true for a channel-wide prefetch.
+	 * @since 2.2.17
+	 * @see Channel#basicQos(int, boolean)
+	 */
+	public void setGlobalQos(boolean globalQos) {
+		this.globalQos = globalQos;
+	}
+
 	/**
 	 * Return true if cancellation is expected.
 	 * @return true if expected.
@@ -608,10 +620,8 @@ private void passiveDeclarations() {
 
 	private void setQosAndreateConsumers() {
 		if (!this.acknowledgeMode.isAutoAck() && !cancelled()) {
-			// Set basicQos before calling basicConsume (otherwise if we are not acking the broker
-			// will send blocks of 100 messages)
 			try {
-				this.channel.basicQos(this.prefetchCount);
+				this.channel.basicQos(this.prefetchCount, this.globalQos);
 			}
 			catch (IOException e) {
 				this.activeObjectCounter.release(this);
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
@@ -724,7 +724,7 @@ private SimpleConsumer consume(String queue, int index, Connection connection) {
 		SimpleConsumer consumer = null;
 		try {
 			channel = connection.createChannel(isChannelTransacted());
-			channel.basicQos(getPrefetchCount());
+			channel.basicQos(getPrefetchCount(), isGlobalQos());
 			consumer = new SimpleConsumer(connection, channel, queue, index);
 			channel.queueDeclarePassive(queue);
 			consumer.consumerTag = channel.basicConsume(queue, getAcknowledgeMode().isAutoAck(),
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
@@ -831,6 +831,7 @@ protected BlockingQueueConsumer createBlockingQueueConsumer() {
 		consumer = new BlockingQueueConsumer(getConnectionFactory(), getMessagePropertiesConverter(),
 				this.cancellationLock, getAcknowledgeMode(), isChannelTransacted(), actualPrefetchCount,
 				isDefaultRequeueRejected(), getConsumerArguments(), isNoLocal(), isExclusive(), queues);
+		consumer.setGlobalQos(isGlobalQos());
 		consumer.setMissingQueuePublisher(this::publishMissingQueueEvent);
 		if (this.declarationRetries != null) {
 			consumer.setDeclarationRetries(this.declarationRetries);
diff --git a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-2.2.xsd b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-2.2.xsd
@@ -728,11 +728,24 @@
 		<xsd:attribute name="prefetch" type="xsd:string">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[
-	Tells the broker how many messages to send to each consumer in a single request. Often this can be set quite high
-	to improve throughput. It should be greater than or equal to the transaction size.
+	Tells the broker how many messages to send to each consumer (or channel) in a single request.
+	Often this can be set quite high
+	to improve throughput, but it can cause starvation when you have multiple application instances
+	and low message volume.
+	It should be greater than or equal to the batch size. Also see 'global-qos'.
 				]]></xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
+		<xsd:attribute name="global-qos">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	When true, apply the 'prefetch' globally to the channel rather than to each consumer on the channel.
+				]]></xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:union memberTypes="xsd:boolean xsd:string" />
+			</xsd:simpleType>
+		</xsd:attribute>
 		<xsd:attribute name="transaction-size" type="xsd:string">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ListenerContainerParserTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ListenerContainerParserTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2020 the original author or authors.
+ * Copyright 2010-2021 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.
@@ -104,6 +104,7 @@ public void testParseWithQueueNames() {
 		assertThat(ReflectionTestUtils.getField(container, "idleEventInterval")).isEqualTo(1235L);
 		assertThat(container.getListenerId()).isEqualTo("container1");
 		assertThat(TestUtils.getPropertyValue(container, "mismatchedQueuesFatal", Boolean.class)).isTrue();
+		assertThat(TestUtils.getPropertyValue(container, "globalQos", Boolean.class)).isFalse();
 	}
 
 	@Test
@@ -149,6 +150,7 @@ public void testParseWithQueues() {
 		assertThat(Arrays.asList(container.getQueueNames()).toString()).isEqualTo("[foo, " + queue.getName() + "]");
 		assertThat(TestUtils.getPropertyValue(container, "missingQueuesFatal", Boolean.class)).isTrue();
 		assertThat(TestUtils.getPropertyValue(container, "autoDeclare", Boolean.class)).isFalse();
+		assertThat(TestUtils.getPropertyValue(container, "globalQos", Boolean.class)).isTrue();
 	}
 
 	@Test
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/RabbitListenerContainerFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/RabbitListenerContainerFactoryTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -103,6 +103,7 @@ public void createContainerFullConfig() {
 		this.factory.setRecoveryBackOff(recoveryBackOff);
 		this.factory.setMissingQueuesFatal(true);
 		this.factory.setAfterReceivePostProcessors(afterReceivePostProcessor);
+		this.factory.setGlobalQos(true);
 		this.factory.setContainerCustomizer(c -> c.setShutdownTimeout(10_000));
 
 		assertThat(this.factory.getAdviceChain()).isEqualTo(new Advice[]{advice});
@@ -138,6 +139,7 @@ public void createContainerFullConfig() {
 		List<?> actualAfterReceivePostProcessors = (List<?>) fieldAccessor.getPropertyValue("afterReceivePostProcessors");
 		assertThat(actualAfterReceivePostProcessors.size()).as("Wrong number of afterReceivePostProcessors").isEqualTo(1);
 		assertThat(actualAfterReceivePostProcessors.get(0)).as("Wrong advice").isSameAs(afterReceivePostProcessor);
+		assertThat(fieldAccessor.getPropertyValue("globalQos")).isEqualTo(true);
 	}
 
 	@Test
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
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -163,7 +163,7 @@ public void testPrefetchIsSetOnFailedPassiveDeclaration() throws IOException {
 		blockingQueueConsumer.setFailedDeclarationRetryInterval(10);
 		blockingQueueConsumer.start();
 
-		verify(channel).basicQos(20);
+		verify(channel).basicQos(20, false);
 	}
 
 	@Test
@@ -293,7 +293,7 @@ public void testAlwaysCancelAutoRecoverConsumer() throws IOException {
 		blockingQueueConsumer.setFailedDeclarationRetryInterval(10);
 		blockingQueueConsumer.start();
 
-		verify(channel).basicQos(2);
+		verify(channel).basicQos(2, false);
 		isOpen.set(false);
 		blockingQueueConsumer.stop();
 		verify(channel).basicCancel("consumerTag");
@@ -335,7 +335,7 @@ public void testDrainAndReject() throws IOException, TimeoutException {
 		blockingQueueConsumer.setFailedDeclarationRetryInterval(10);
 		blockingQueueConsumer.start();
 
-		verify(channel).basicQos(2);
+		verify(channel).basicQos(2, false);
 		Consumer consumer = (Consumer) TestUtils.getPropertyValue(blockingQueueConsumer, "consumers", Map.class)
 				.get("test");
 		isOpen.set(false);
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
@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2021 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.
@@ -86,7 +86,7 @@ public void testAlwaysCancelAutoRecoverConsumer() throws Exception {
 			qos.set(i.getArgument(0));
 			latch1.countDown();
 			return null;
-		}).given(channel).basicQos(anyInt());
+		}).given(channel).basicQos(anyInt(), anyBoolean());
 		final CountDownLatch latch2 = new CountDownLatch(1);
 		willAnswer(i -> {
 			latch2.countDown();
@@ -135,7 +135,7 @@ public void testDeferredAcks() throws Exception {
 		willAnswer(i -> {
 			qos.set(i.getArgument(0));
 			return null;
-		}).given(channel).basicQos(anyInt());
+		}).given(channel).basicQos(anyInt(), anyBoolean());
 		final CountDownLatch latch2 = new CountDownLatch(2);
 		final CountDownLatch latch3 = new CountDownLatch(1);
 		willAnswer(i -> {
@@ -233,7 +233,7 @@ public void testRemoveQueuesWhileNotConnected() throws Exception {
 			qos.set(i.getArgument(0));
 			latch1.countDown();
 			return null;
-		}).given(channel).basicQos(anyInt());
+		}).given(channel).basicQos(anyInt(), anyBoolean());
 		final CountDownLatch latch2 = new CountDownLatch(2);
 		willAnswer(i -> {
 			latch2.countDown();
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java
@@ -433,18 +433,21 @@ public void testRestartConsumerOnBasicQosIoException() throws Exception {
 
 		final AtomicBoolean networkGlitch = new AtomicBoolean();
 
+		final AtomicBoolean globalQos = new AtomicBoolean();
+
 		class MockChannel extends PublisherCallbackChannelImpl {
 
 			MockChannel(Channel delegate) {
 				super(delegate, SimpleMessageListenerContainerIntegration2Tests.this.executorService);
 			}
 
 			@Override
-			public void basicQos(int prefetchCount) throws IOException {
+			public void basicQos(int prefetchCount, boolean global) throws IOException {
+				globalQos.set(global);
 				if (networkGlitch.compareAndSet(false, true)) {
 					throw new IOException("Intentional connection reset");
 				}
-				super.basicQos(prefetchCount);
+				super.basicQos(prefetchCount, global);
 			}
 
 		}
@@ -461,11 +464,13 @@ public void basicQos(int prefetchCount) throws IOException {
 		container.setMessageListener(new MessageListenerAdapter(new PojoListener(latch)));
 		container.setQueueNames(queue.getName());
 		container.setRecoveryInterval(500);
+		container.setGlobalQos(true);
 		container.afterPropertiesSet();
 		container.start();
 
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(networkGlitch.get()).isTrue();
+		assertThat(globalQos.get()).isTrue();
 
 		container.stop();
 		((DisposableBean) connectionFactory).destroy();
diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ListenerContainerParserTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ListenerContainerParserTests-context.xml
@@ -24,7 +24,7 @@
 	</rabbit:listener-container>
 
 	<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" concurrency="5"
-			auto-declare="false">
+			auto-declare="false" global-qos="true">
 		<rabbit:listener id="container2" queues="foo, bar" ref="testBean" method="handle"/>
 	</rabbit:listener-container>
 
diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc

Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,5 @@`
`1`	`1`	`/*`
`2`		`- * Copyright 2010-2020 the original author or authors.`
	`2`	`+ * Copyright 2010-2021 the original author or authors.`
`3`	`3`	`*`
`4`	`4`	`* Licensed under the Apache License, Version 2.0 (the "License");`
`5`	`5`	`* you may not use this file except in compliance with the License.`
`@@ -104,6 +104,7 @@ public void testParseWithQueueNames() {`
`104`	`104`	`assertThat(ReflectionTestUtils.getField(container, "idleEventInterval")).isEqualTo(1235L);`
`105`	`105`	`assertThat(container.getListenerId()).isEqualTo("container1");`
`106`	`106`	`assertThat(TestUtils.getPropertyValue(container, "mismatchedQueuesFatal", Boolean.class)).isTrue();`
	`107`	`+ assertThat(TestUtils.getPropertyValue(container, "globalQos", Boolean.class)).isFalse();`
`107`	`108`	`}`
`108`	`109`
`109`	`110`	`@Test`
`@@ -149,6 +150,7 @@ public void testParseWithQueues() {`
`149`	`150`	`assertThat(Arrays.asList(container.getQueueNames()).toString()).isEqualTo("[foo, " + queue.getName() + "]");`
`150`	`151`	`assertThat(TestUtils.getPropertyValue(container, "missingQueuesFatal", Boolean.class)).isTrue();`
`151`	`152`	`assertThat(TestUtils.getPropertyValue(container, "autoDeclare", Boolean.class)).isFalse();`
	`153`	`+ assertThat(TestUtils.getPropertyValue(container, "globalQos", Boolean.class)).isTrue();`
`152`	`154`	`}`
`153`	`155`
`154`	`156`	`@Test`