Add configuration options for RabbitMQ's batch listener config

See gh-23766

Author: garyrussell

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/AbstractRabbitListenerContainerFactoryConfigurer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2019 the original author or authors.
+ * Copyright 2012-2020 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.
@@ -116,6 +116,7 @@ protected void configure(T factory, ConnectionFactory connectionFactory,
 			factory.setIdleEventInterval(configuration.getIdleEventInterval().toMillis());
 		}
 		factory.setMissingQueuesFatal(configuration.isMissingQueuesFatal());
+		factory.setDeBatchingEnabled(configuration.isDeBatchingEnabled());
 		ListenerRetry retryConfig = configuration.getRetry();
 		if (retryConfig.isEnabled()) {
 			RetryInterceptorBuilder<?, ?> builder = (retryConfig.isStateless()) ? RetryInterceptorBuilder.stateless()

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitProperties.java

@@ -662,6 +662,11 @@ public abstract static class AmqpContainer {
 		 */
 		private Duration idleEventInterval;
 
+		/**
+		 * Whether to present batched messages (created by a BatchingRabbitTemplate) as discrete messages.
+		 */
+		private boolean deBatchingEnabled = true;
+
 		/**
 		 * Optional properties for a retry interceptor.
 		 */
@@ -709,6 +714,14 @@ public void setIdleEventInterval(Duration idleEventInterval) {
 
 		public abstract boolean isMissingQueuesFatal();
 
+		public boolean isDeBatchingEnabled() {
+			return deBatchingEnabled;
+		}
+
+		public void setDeBatchingEnabled(boolean deBatchingEnabled) {
+			this.deBatchingEnabled = deBatchingEnabled;
+		}
+
 		public ListenerRetry getRetry() {
 			return this.retry;
 		}
@@ -743,6 +756,13 @@ public static class SimpleContainer extends AmqpContainer {
 		 */
 		private boolean missingQueuesFatal = true;
 
+		/**
+		 * When true, the container will create a batch of messages based on the 'receiveTimeout' and 'batchSize'.
+		 * Coerces 'deBatchingEnabled' to true to include the contents of a producer created batch in the batch as
+		 * discrete records.
+		 */
+		private boolean consumerBatchEnabled;
+
 		public Integer getConcurrency() {
 			return this.concurrency;
 		}
@@ -776,6 +796,14 @@ public void setMissingQueuesFatal(boolean missingQueuesFatal) {
 			this.missingQueuesFatal = missingQueuesFatal;
 		}
 
+		public boolean isConsumerBatchEnabled() {
+			return consumerBatchEnabled;
+		}
+
+		public void setConsumerBatchEnabled(boolean consumerBatchEnabled) {
+			this.consumerBatchEnabled = consumerBatchEnabled;
+		}
+
 	}
 
 	/**

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/SimpleRabbitListenerContainerFactoryConfigurer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2019 the original author or authors.
+ * Copyright 2012-2020 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.
@@ -39,6 +39,10 @@ public void configure(SimpleRabbitListenerContainerFactory factory, ConnectionFa
 		map.from(config::getConcurrency).whenNonNull().to(factory::setConcurrentConsumers);
 		map.from(config::getMaxConcurrency).whenNonNull().to(factory::setMaxConcurrentConsumers);
 		map.from(config::getBatchSize).whenNonNull().to(factory::setBatchSize);
+		map.from(config::isConsumerBatchEnabled).to(factory::setConsumerBatchEnabled);
+		if (config.isConsumerBatchEnabled()) {
+			factory.setDeBatchingEnabled(true);
+		}
 	}
 
 }

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfigurationTests.java

@@ -34,6 +34,7 @@
 import org.aopalliance.aop.Advice;
 import org.junit.jupiter.api.Test;
 
+import org.mockito.InOrder;
 import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.AmqpAdmin;
 import org.springframework.amqp.core.Message;
@@ -75,8 +76,7 @@
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.isNull;
 import static org.mockito.BDDMockito.given;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.*;
 
 /**
  * Tests for {@link RabbitAutoConfiguration}.
@@ -538,15 +538,21 @@ void testSimpleRabbitListenerContainerFactoryConfigurerUsesConfig() {
 				.withPropertyValues("spring.rabbitmq.listener.type:direct",
 						"spring.rabbitmq.listener.simple.concurrency:5",
 						"spring.rabbitmq.listener.simple.maxConcurrency:10",
-						"spring.rabbitmq.listener.simple.prefetch:40")
+						"spring.rabbitmq.listener.simple.prefetch:40",
+						"spring.rabbitmq.listener.simple.consumer-batch-enabled:true",
+						"spring.rabbitmq.listener.simple.de-batching-enabled:false")
 				.run((context) -> {
 					SimpleRabbitListenerContainerFactoryConfigurer configurer = context
 							.getBean(SimpleRabbitListenerContainerFactoryConfigurer.class);
 					SimpleRabbitListenerContainerFactory factory = mock(SimpleRabbitListenerContainerFactory.class);
 					configurer.configure(factory, mock(ConnectionFactory.class));
+					InOrder inOrder = inOrder(factory);
 					verify(factory).setConcurrentConsumers(5);
 					verify(factory).setMaxConcurrentConsumers(10);
 					verify(factory).setPrefetchCount(40);
+					verify(factory).setConsumerBatchEnabled(true);
+					inOrder.verify(factory).setDeBatchingEnabled(false);
+					inOrder.verify(factory).setDeBatchingEnabled(true);
 				});
 	}
 
@@ -555,14 +561,16 @@ void testDirectRabbitListenerContainerFactoryConfigurerUsesConfig() {
 		this.contextRunner.withUserConfiguration(TestConfiguration.class)
 				.withPropertyValues("spring.rabbitmq.listener.type:simple",
 						"spring.rabbitmq.listener.direct.consumers-per-queue:5",
-						"spring.rabbitmq.listener.direct.prefetch:40")
+						"spring.rabbitmq.listener.direct.prefetch:40",
+						"spring.rabbitmq.listener.direct.de-batching-enabled:false")
 				.run((context) -> {
 					DirectRabbitListenerContainerFactoryConfigurer configurer = context
 							.getBean(DirectRabbitListenerContainerFactoryConfigurer.class);
 					DirectRabbitListenerContainerFactory factory = mock(DirectRabbitListenerContainerFactory.class);
 					configurer.configure(factory, mock(ConnectionFactory.class));
 					verify(factory).setConsumersPerQueue(5);
 					verify(factory).setPrefetchCount(40);
+					verify(factory).setDeBatchingEnabled(false);
 				});
 	}
 

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/amqp/RabbitPropertiesTests.java

@@ -303,6 +303,8 @@ void simpleContainerUseConsistentDefaultValues() {
 		RabbitProperties.SimpleContainer simple = this.properties.getListener().getSimple();
 		assertThat(simple.isAutoStartup()).isEqualTo(container.isAutoStartup());
 		assertThat(container).hasFieldOrPropertyWithValue("missingQueuesFatal", simple.isMissingQueuesFatal());
+		assertThat(container).hasFieldOrPropertyWithValue("deBatchingEnabled", simple.isDeBatchingEnabled());
+		assertThat(container).hasFieldOrPropertyWithValue("consumerBatchEnabled", simple.isConsumerBatchEnabled());
 	}
 
 	@Test
@@ -312,6 +314,7 @@ void directContainerUseConsistentDefaultValues() {
 		RabbitProperties.DirectContainer direct = this.properties.getListener().getDirect();
 		assertThat(direct.isAutoStartup()).isEqualTo(container.isAutoStartup());
 		assertThat(container).hasFieldOrPropertyWithValue("missingQueuesFatal", direct.isMissingQueuesFatal());
+		assertThat(container).hasFieldOrPropertyWithValue("deBatchingEnabled", direct.isDeBatchingEnabled());
 	}
 
 }