GH-1318: Support the Global Flag in basicQos

Resolves https://github.com/spring-projects/spring-amqp/issues/1318

**cherry-pick to 2.2.x**

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.
@@ -86,6 +86,8 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 
 	private Integer prefetchCount;
 
+	private Boolean globalQos;
+
 	private Boolean defaultRequeueRejected;
 
 	private Advice[] adviceChain;
@@ -387,6 +389,16 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 		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();
@@ -405,6 +417,7 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 			.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)

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 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.transaction.interceptor.TransactionAttribute;
 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 class ListenerContainerFactoryBean extends AbstractFactoryBean<AbstractMe
 		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;
 	}
@@ -449,6 +463,7 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean<AbstractMe
 					.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)

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 final class RabbitNamespaceUtils {
 			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));

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java

@@ -220,6 +220,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 
 	private int prefetchCount = DEFAULT_PREFETCH_COUNT;
 
+	private boolean globalQos;
+
 	private long idleEventInterval;
 
 	private long lastReceive = System.currentTimeMillis();
@@ -797,6 +799,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 	 * 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;
@@ -811,6 +814,20 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 		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

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java

@@ -162,6 +162,8 @@ public class BlockingQueueConsumer {
 
 	private java.util.function.Consumer<String> missingQueuePublisher = str -> { };
 
+	private boolean globalQos;
+
 	private volatile long abortStarted;
 
 	private volatile boolean normalCancel;
@@ -405,6 +407,16 @@ public class BlockingQueueConsumer {
 		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.
@@ -629,10 +641,8 @@ public class BlockingQueueConsumer {
 			}
 		}
 		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);

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java

@@ -736,7 +736,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 				}
 			}
 			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(),

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java

@@ -822,6 +822,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 		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);

spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit.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[

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 class ListenerContainerParserTests {
 		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 class ListenerContainerParserTests {
 		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

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 class RabbitListenerContainerFactoryTests {
 		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 class RabbitListenerContainerFactoryTests {
 		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

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.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.
@@ -161,7 +161,7 @@ public class BlockingQueueConsumerTests {
 		blockingQueueConsumer.setFailedDeclarationRetryInterval(10);
 		blockingQueueConsumer.start();
 
-		verify(channel).basicQos(20);
+		verify(channel).basicQos(20, false);
 	}
 
 	@Test
@@ -291,7 +291,7 @@ public class BlockingQueueConsumerTests {
 		blockingQueueConsumer.setFailedDeclarationRetryInterval(10);
 		blockingQueueConsumer.start();
 
-		verify(channel).basicQos(2);
+		verify(channel).basicQos(2, false);
 		isOpen.set(false);
 		blockingQueueConsumer.stop();
 		verify(channel).basicCancel("consumerTag");
@@ -333,7 +333,7 @@ public class BlockingQueueConsumerTests {
 		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);

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 class DirectMessageListenerContainerMockTests {
 			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 class DirectMessageListenerContainerMockTests {
 		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 class DirectMessageListenerContainerMockTests {
 			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();

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java

@@ -424,6 +424,8 @@ public class SimpleMessageListenerContainerIntegration2Tests {
 
 		final AtomicBoolean networkGlitch = new AtomicBoolean();
 
+		final AtomicBoolean globalQos = new AtomicBoolean();
+
 		class MockChannel extends PublisherCallbackChannelImpl {
 
 			MockChannel(Channel delegate) {
@@ -431,11 +433,12 @@ public class SimpleMessageListenerContainerIntegration2Tests {
 			}
 
 			@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);
 			}
 
 		}
@@ -452,11 +455,13 @@ public class SimpleMessageListenerContainerIntegration2Tests {
 		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();

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>
 

src/reference/asciidoc/amqp.adoc

@@ -5672,6 +5672,15 @@ You can set it to `false` to revert to the previous behavior.
 a|image::images/tickmark.png[]
 a|image::images/tickmark.png[]
 
+|globalQos
+(global-qos)
+
+|When true, the `prefetchCount` is applied globally to the channel rather than to each consumer on the channel.
+See https://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.qos.global[`basicQos.global`] for more information.
+
+a|image::images/tickmark.png[]
+a|image::images/tickmark.png[]
+
 |(group)
 
 |This is available only when using the namespace.
@@ -5866,6 +5875,7 @@ to a large amount of memory in the client process), and if strict message orderi
 (the prefetch value should be set back to 1 in this case).
 Also, with low-volume messaging and multiple consumers (including concurrency within a single listener container instance), you may wish to reduce the prefetch to get a more even distribution of messages across consumers.
 
+Also see `globalQos`.
 
 a|image::images/tickmark.png[]
 a|image::images/tickmark.png[]

src/reference/asciidoc/whats-new.adoc

@@ -41,6 +41,8 @@ A new listener container property `consumeDelay` is now available; it is helpful
 The default `JavaLangErrorHandler` now calls `System.exit(99)`.
 To revert to the previous behavior (do nothing), add a no-op handler.
 
+The containers now support the `globalQos` property to apply the `prefetchCount` globally for the channel rather than for each consumer on the channel.
+
 See <<containerAttributes>> for more information.
 
 ==== MessagePostProcessor Changes