GH-3001: Add consume batch support to RabbitAmqpListenerContainer

Fixes: https://github.com/spring-projects/spring-amqp/issues/3001

* Expose batch related options for `RabbitAmqpListenerContainer` and `RabbitAmqpListenerContainerFactory`, respectively
* `batchSize` - the indicator that `RabbitAmqpListenerContainer` (and `RabbitAmqpMessageListenerAdapter`) has to work in batch mode
* `batchReceiveTimeout` - how long to wait for batch to be fulfilled or release whatever was gathered so far, even if just only one message
* `taskScheduler` - schedule "force batch release" after `batchReceiveTimeout`
* Make `MessagingMessageListenerAdapter` `final` properties as `protected` to avoid undesired copy-paste burden

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

@@ -49,8 +49,6 @@ import org.springframework.util.Assert;
 public class BatchMessagingMessageListenerAdapter extends MessagingMessageListenerAdapter
 		implements ChannelAwareBatchMessageListener {
 
-	private final MessagingMessageConverterAdapter converterAdapter;
-
 	private final BatchingStrategy batchingStrategy;
 
 	@SuppressWarnings("this-escape")
@@ -58,14 +56,13 @@ public class BatchMessagingMessageListenerAdapter extends MessagingMessageListen
 			@Nullable RabbitListenerErrorHandler errorHandler, @Nullable BatchingStrategy batchingStrategy) {
 
 		super(bean, method, returnExceptions, errorHandler, true);
-		this.converterAdapter = (MessagingMessageConverterAdapter) getMessagingMessageConverter();
 		this.batchingStrategy = batchingStrategy == null ? new SimpleBatchingStrategy(0, 0, 0L) : batchingStrategy;
 	}
 
 	@Override
 	public void onMessageBatch(List<org.springframework.amqp.core.Message> messages, @Nullable Channel channel) {
 		Message<?> converted;
-		if (this.converterAdapter.isAmqpMessageList()) {
+		if (this.messagingMessageConverter.isAmqpMessageList()) {
 			converted = new GenericMessage<>(messages);
 		}
 		else {
@@ -87,7 +84,7 @@ public class BatchMessagingMessageListenerAdapter extends MessagingMessageListen
 					}
 				}
 			}
-			if (this.converterAdapter.isMessageList()) {
+			if (this.messagingMessageConverter.isMessageList()) {
 				converted = new GenericMessage<>(messagingMessages);
 			}
 			else {
@@ -178,7 +175,7 @@ public class BatchMessagingMessageListenerAdapter extends MessagingMessageListen
 	protected Message<?> toMessagingMessage(org.springframework.amqp.core.Message amqpMessage) {
 		if (this.batchingStrategy.canDebatch(amqpMessage.getMessageProperties())) {
 
-			if (this.converterAdapter.isMessageList()) {
+			if (this.messagingMessageConverter.isMessageList()) {
 				List<Message<?>> messages = new ArrayList<>();
 				this.batchingStrategy.deBatch(amqpMessage, fragment -> messages.add(super.toMessagingMessage(fragment)));
 				return new GenericMessage<>(messages);
@@ -186,9 +183,9 @@ public class BatchMessagingMessageListenerAdapter extends MessagingMessageListen
 			else {
 				List<Object> list = new ArrayList<>();
 				this.batchingStrategy.deBatch(amqpMessage, fragment ->
-						list.add(this.converterAdapter.extractPayload(fragment)));
+						list.add(this.messagingMessageConverter.extractPayload(fragment)));
 				return MessageBuilder.withPayload(list)
-						.copyHeaders(this.converterAdapter
+						.copyHeaders(this.messagingMessageConverter
 								.getHeaderMapper()
 								.toHeaders(amqpMessage.getMessageProperties()))
 						.build();

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

@@ -69,11 +69,11 @@ import org.springframework.util.TypeUtils;
  */
 public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageListener {
 
-	private final MessagingMessageConverterAdapter messagingMessageConverter;
+	protected final MessagingMessageConverterAdapter messagingMessageConverter;
 
-	private final boolean returnExceptions;
+	protected final boolean returnExceptions;
 
-	private final @Nullable RabbitListenerErrorHandler errorHandler;
+	protected final @Nullable RabbitListenerErrorHandler errorHandler;
 
 	private @Nullable HandlerAdapter handlerAdapter;
 
@@ -364,15 +364,15 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
 			}
 		}
 
-		protected boolean isMessageList() {
+		public boolean isMessageList() {
 			return this.isMessageList;
 		}
 
-		protected boolean isAmqpMessageList() {
+		public boolean isAmqpMessageList() {
 			return this.isAmqpMessageList;
 		}
 
-		protected @Nullable Method getMethod() {
+		public @Nullable Method getMethod() {
 			return this.method;
 		}
 

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/config/RabbitAmqpListenerContainerFactory.java

@@ -16,10 +16,13 @@
 
 package org.springframework.amqp.rabbitmq.client.config;
 
+import java.util.Arrays;
+
 import com.rabbitmq.client.amqp.Connection;
-import org.aopalliance.aop.Advice;
 import org.jspecify.annotations.Nullable;
 
+import org.springframework.amqp.core.MessageListener;
+import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.rabbit.config.BaseRabbitListenerContainerFactory;
 import org.springframework.amqp.rabbit.config.ContainerCustomizer;
 import org.springframework.amqp.rabbit.listener.MethodRabbitListenerEndpoint;
@@ -27,6 +30,7 @@ import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
 import org.springframework.amqp.rabbitmq.client.listener.RabbitAmqpListenerContainer;
 import org.springframework.amqp.rabbitmq.client.listener.RabbitAmqpMessageListenerAdapter;
 import org.springframework.amqp.utils.JavaUtils;
+import org.springframework.scheduling.TaskScheduler;
 
 /**
  * Factory for {@link RabbitAmqpListenerContainer}.
@@ -45,6 +49,14 @@ public class RabbitAmqpListenerContainerFactory
 
 	private @Nullable ContainerCustomizer<RabbitAmqpListenerContainer> containerCustomizer;
 
+	private MessagePostProcessor @Nullable [] afterReceivePostProcessors;
+
+	private @Nullable Integer batchSize;
+
+	private @Nullable Long batchReceiveTimeout;
+
+	private @Nullable TaskScheduler taskScheduler;
+
 	/**
 	 * Construct an instance using the provided amqpConnection.
 	 * @param amqpConnection the connection.
@@ -62,18 +74,69 @@ public class RabbitAmqpListenerContainerFactory
 		this.containerCustomizer = containerCustomizer;
 	}
 
+	/**
+	 * Set {@link MessagePostProcessor}s that will be applied after message reception, before
+	 * invoking the {@link MessageListener}. Often used to decompress data.  Processors are invoked in order,
+	 * depending on {@code PriorityOrder}, {@code Order} and finally unordered.
+	 * @param afterReceivePostProcessors the post processor.
+	 */
+	public void setAfterReceivePostProcessors(MessagePostProcessor... afterReceivePostProcessors) {
+		this.afterReceivePostProcessors = Arrays.copyOf(afterReceivePostProcessors, afterReceivePostProcessors.length);
+	}
+
+	/**
+	 * The size of the batch of messages to process.
+	 * This is only option (if {@code batchSize > 1}) which turns the target listener container into a batch mode.
+	 * @param batchSize the batch size.
+	 * @see RabbitAmqpListenerContainer#setBatchSize
+	 * @see #setBatchReceiveTimeout(Long)
+	 */
+	public void setBatchSize(Integer batchSize) {
+		this.batchSize = batchSize;
+	}
+
+	/**
+	 * The number of milliseconds of timeout for gathering batch messages.
+	 * It limits the time to wait to fill batchSize.
+	 * Default is 30 seconds.
+	 * @param batchReceiveTimeout the timeout for gathering batch messages.
+	 * @see RabbitAmqpListenerContainer#setBatchReceiveTimeout
+	 * @see #setBatchSize(Integer)
+	 */
+	public void setBatchReceiveTimeout(Long batchReceiveTimeout) {
+		this.batchReceiveTimeout = batchReceiveTimeout;
+	}
+
+	/**
+	 * Configure a {@link TaskScheduler} to release not fulfilled batches after timeout.
+	 * @param taskScheduler the {@link TaskScheduler} to use.
+	 * @see RabbitAmqpListenerContainer#setTaskScheduler(TaskScheduler)
+	 * @see #setBatchReceiveTimeout(Long)
+	 */
+	public void setTaskScheduler(TaskScheduler taskScheduler) {
+		this.taskScheduler = taskScheduler;
+	}
+
 	@Override
 	public RabbitAmqpListenerContainer createListenerContainer(@Nullable RabbitListenerEndpoint endpoint) {
 		if (endpoint instanceof MethodRabbitListenerEndpoint methodRabbitListenerEndpoint) {
+			JavaUtils.INSTANCE
+					.acceptIfCondition(this.batchSize != null && this.batchSize > 1,
+							true,
+							methodRabbitListenerEndpoint::setBatchListener);
+
 			methodRabbitListenerEndpoint.setAdapterProvider(
 					(batch, bean, method, returnExceptions, errorHandler, batchingStrategy) ->
-							new RabbitAmqpMessageListenerAdapter(bean, method, returnExceptions, errorHandler));
+							new RabbitAmqpMessageListenerAdapter(bean, method, returnExceptions, errorHandler, batch));
 		}
 		RabbitAmqpListenerContainer container = createContainerInstance();
-		Advice[] adviceChain = getAdviceChain();
 		JavaUtils.INSTANCE
-				.acceptIfNotNull(adviceChain, container::setAdviceChain)
-				.acceptIfNotNull(getDefaultRequeueRejected(), container::setDefaultRequeue);
+				.acceptIfNotNull(getAdviceChain(), container::setAdviceChain)
+				.acceptIfNotNull(getDefaultRequeueRejected(), container::setDefaultRequeue)
+				.acceptIfNotNull(this.afterReceivePostProcessors, container::setAfterReceivePostProcessors)
+				.acceptIfNotNull(this.batchSize, container::setBatchSize)
+				.acceptIfNotNull(this.batchReceiveTimeout, container::setBatchReceiveTimeout)
+				.acceptIfNotNull(this.taskScheduler, container::setTaskScheduler);
 
 		applyCommonOverrides(endpoint, container);
 

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerContainer.java

@@ -17,9 +17,14 @@
 package org.springframework.amqp.rabbitmq.client.listener;
 
 import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
@@ -36,13 +41,19 @@ import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.AmqpAcknowledgment;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageListener;
+import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
 import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.support.ContainerUtils;
 import org.springframework.amqp.rabbitmq.client.RabbitAmqpUtils;
+import org.springframework.amqp.support.postprocessor.MessagePostProcessorUtils;
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.aop.support.DefaultPointcutAdvisor;
+import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.core.log.LogAccessor;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.util.Assert;
 import org.springframework.util.ErrorHandler;
 import org.springframework.util.LinkedMultiValueMap;
@@ -57,7 +68,7 @@ import org.springframework.util.ObjectUtils;
  * @since 4.0
  *
  */
-public class RabbitAmqpListenerContainer implements MessageListenerContainer {
+public class RabbitAmqpListenerContainer implements MessageListenerContainer, BeanNameAware, DisposableBean {
 
 	private static final LogAccessor LOG = new LogAccessor(LogFactory.getLog(RabbitAmqpListenerContainer.class));
 
@@ -85,14 +96,28 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer {
 
 	private @Nullable MessageListener messageListener;
 
+	private @Nullable MessageListener proxy;
+
 	private ErrorHandler errorHandler = new ConditionalRejectingErrorHandler();
 
+	private @Nullable Collection<MessagePostProcessor> afterReceivePostProcessors;
+
 	private boolean autoStartup = true;
 
+	private String beanName = "not.a.Spring.bean";
+
 	private @Nullable String listenerId;
 
 	private Duration gracefulShutdownPeriod = Duration.ofSeconds(30);
 
+	private int batchSize;
+
+	private Duration batchReceiveDuration = Duration.ofSeconds(30);
+
+	private @Nullable TaskScheduler taskScheduler;
+
+	private boolean internalTaskScheduler = true;
+
 	/**
 	 * Construct an instance using the provided connection.
 	 * @param connection to use.
@@ -118,6 +143,39 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer {
 		this.stateListeners = Arrays.copyOf(stateListeners, stateListeners.length);
 	}
 
+	/**
+	 * Set {@link MessagePostProcessor}s that will be applied after message reception, before
+	 * invoking the {@link MessageListener}. Often used to decompress data.  Processors are invoked in order,
+	 * depending on {@code PriorityOrder}, {@code Order} and finally unordered.
+	 * @param afterReceivePostProcessors the post processor.
+	 */
+	public void setAfterReceivePostProcessors(MessagePostProcessor... afterReceivePostProcessors) {
+		this.afterReceivePostProcessors = MessagePostProcessorUtils.sort(Arrays.asList(afterReceivePostProcessors));
+	}
+
+	public void setBatchSize(int batchSize) {
+		Assert.isTrue(batchSize > 1, "'batchSize' must be greater than 1");
+		this.batchSize = batchSize;
+	}
+
+	public void setBatchReceiveTimeout(long batchReceiveTimeout) {
+		this.batchReceiveDuration = Duration.ofMillis(batchReceiveTimeout);
+	}
+
+	/**
+	 * Set a {@link TaskScheduler} for monitoring batch releases.
+	 * @param taskScheduler the {@link TaskScheduler} to use.
+	 */
+	public void setTaskScheduler(TaskScheduler taskScheduler) {
+		this.taskScheduler = taskScheduler;
+		this.internalTaskScheduler = false;
+	}
+
+	@Override
+	public void setBeanName(String name) {
+		this.beanName = name;
+	}
+
 	@Override
 	public void setAutoStartup(boolean autoStart) {
 		this.autoStartup = autoStart;
@@ -185,22 +243,31 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer {
 		this.listenerId = id;
 	}
 
+	/**
+	 * The 'id' attribute of the listener.
+	 * @return the id (or the container bean name if no id set).
+	 */
+	public String getListenerId() {
+		return this.listenerId != null ? this.listenerId : this.beanName;
+	}
+
 	@Override
 	public void setupMessageListener(MessageListener messageListener) {
 		this.messageListener = messageListener;
+		this.proxy = this.messageListener;
 		if (!ObjectUtils.isEmpty(this.adviceChain)) {
 			ProxyFactory factory = new ProxyFactory(messageListener);
 			for (Advice advice : this.adviceChain) {
 				factory.addAdvisor(new DefaultPointcutAdvisor(advice));
 			}
 			factory.setInterfaces(messageListener.getClass().getInterfaces());
-			this.messageListener = (MessageListener) factory.getProxy(getClass().getClassLoader());
+			this.proxy = (MessageListener) factory.getProxy(getClass().getClassLoader());
 		}
 	}
 
 	@Override
 	public @Nullable Object getMessageListener() {
-		return this.messageListener;
+		return this.proxy;
 	}
 
 	@Override
@@ -209,6 +276,18 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer {
 		Assert.state(this.messageListener != null, "The 'messageListener' must be provided.");
 
 		this.messageListener.containerAckMode(this.autoSettle ? AcknowledgeMode.AUTO : AcknowledgeMode.MANUAL);
+		if (this.messageListener instanceof RabbitAmqpMessageListenerAdapter adapter
+				&& this.afterReceivePostProcessors != null) {
+
+			adapter.setAfterReceivePostProcessors(this.afterReceivePostProcessors);
+		}
+
+		if (this.batchSize > 1 && this.internalTaskScheduler) {
+			ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
+			threadPoolTaskScheduler.setThreadNamePrefix(getListenerId() + "-consumerMonitor-");
+			threadPoolTaskScheduler.afterPropertiesSet();
+			this.taskScheduler = threadPoolTaskScheduler;
+		}
 	}
 
 	@Override
@@ -236,7 +315,7 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer {
 										.priority(this.priority)
 										.initialCredits(this.initialCredits)
 										.listeners(this.stateListeners)
-										.messageHandler(this::invokeListener)
+										.messageHandler(new ConsumerMessageHandler())
 										.build();
 						this.queueToConsumers.add(queue, consumer);
 					}
@@ -256,39 +335,65 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer {
 			}
 		}
 		catch (Exception ex) {
-			try {
-				this.errorHandler.handleError(ex);
-				// If error handler does not re-throw an exception, re-check original error.
-				// If it is not special, treat the error handler outcome as a successful processing result.
-				if (!handleSpecialErrors(ex, context)) {
-					context.accept();
-				}
-			}
-			catch (Exception rethrow) {
-				if (!handleSpecialErrors(rethrow, context)) {
-					if (this.defaultRequeue) {
-						context.requeue();
-					}
-					else {
-						context.discard();
-					}
-					LOG.error(rethrow, () ->
-							"The 'errorHandler' has thrown an exception. The '" + amqpMessage + "' is "
-									+ (this.defaultRequeue ? "re-queued." : "discarded."));
-				}
-			}
+			handleListenerError(ex, context, amqpMessage);
 		}
 	}
 
 	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	private void doInvokeListener(Consumer.Context context, com.rabbitmq.client.amqp.Message amqpMessage) {
 		Consumer.@Nullable Context contextToUse = this.autoSettle ? null : context;
-		if (this.messageListener instanceof RabbitAmqpMessageListener amqpMessageListener) {
+		if (this.proxy instanceof RabbitAmqpMessageListener amqpMessageListener) {
 			amqpMessageListener.onAmqpMessage(amqpMessage, contextToUse);
 		}
 		else {
 			Message message = RabbitAmqpUtils.fromAmqpMessage(amqpMessage, contextToUse);
-			this.messageListener.onMessage(message);
+			this.proxy.onMessage(message);
+		}
+	}
+
+	private void invokeBatchListener(Consumer.Context context, List<com.rabbitmq.client.amqp.Message> batch) {
+		Consumer.@Nullable Context contextToUse = this.autoSettle ? null : context;
+		List<Message> messages =
+				batch.stream()
+						.map((amqpMessage) -> RabbitAmqpUtils.fromAmqpMessage(amqpMessage, contextToUse))
+						.toList();
+		try {
+			doInvokeBatchListener(messages);
+			if (this.autoSettle) {
+				context.accept();
+			}
+		}
+		catch (Exception ex) {
+			handleListenerError(ex, context, batch);
+		}
+	}
+
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
+	private void doInvokeBatchListener(List<Message> messages) {
+		this.proxy.onMessageBatch(messages);
+	}
+
+	private void handleListenerError(Exception ex, Consumer.Context context, Object messageOrBatch) {
+		try {
+			this.errorHandler.handleError(ex);
+			// If error handler does not re-throw an exception, re-check original error.
+			// If it is not special, treat the error handler outcome as a successful processing result.
+			if (!handleSpecialErrors(ex, context)) {
+				context.accept();
+			}
+		}
+		catch (Exception rethrow) {
+			if (!handleSpecialErrors(rethrow, context)) {
+				if (this.defaultRequeue) {
+					context.requeue();
+				}
+				else {
+					context.discard();
+				}
+				LOG.error(rethrow, () ->
+						"The 'errorHandler' has thrown an exception. The '" + messageOrBatch + "' is "
+								+ (this.defaultRequeue ? "re-queued." : "discarded."));
+			}
 		}
 	}
 
@@ -390,4 +495,79 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer {
 		}
 	}
 
+	@Override
+	public void destroy() {
+		if (this.internalTaskScheduler && this.taskScheduler != null) {
+			((ThreadPoolTaskScheduler) this.taskScheduler).shutdown();
+		}
+	}
+
+	private class ConsumerMessageHandler implements Consumer.MessageHandler {
+
+		private volatile @Nullable ConsumerBatch consumerBatch;
+
+		ConsumerMessageHandler() {
+		}
+
+		@Override
+		public void handle(Consumer.Context context, com.rabbitmq.client.amqp.Message message) {
+			if (RabbitAmqpListenerContainer.this.batchSize > 1) {
+				ConsumerBatch currentBatch = this.consumerBatch;
+				if (currentBatch == null || currentBatch.batchReleaseFuture == null) {
+					currentBatch = new ConsumerBatch(context.batch(RabbitAmqpListenerContainer.this.batchSize));
+					this.consumerBatch = currentBatch;
+				}
+				currentBatch.add(context, message);
+				if (currentBatch.batchContext.size() == RabbitAmqpListenerContainer.this.batchSize) {
+					currentBatch.release();
+					this.consumerBatch = null;
+				}
+			}
+			else {
+				invokeListener(context, message);
+			}
+		}
+
+		private class ConsumerBatch {
+
+			private final List<com.rabbitmq.client.amqp.Message> batch = new ArrayList<>();
+
+			private final Consumer.BatchContext batchContext;
+
+			private volatile @Nullable ScheduledFuture<?> batchReleaseFuture;
+
+			ConsumerBatch(Consumer.BatchContext batchContext) {
+				this.batchContext = batchContext;
+			}
+
+			void add(Consumer.Context context, com.rabbitmq.client.amqp.Message message) {
+				this.batchContext.add(context);
+				this.batch.add(message);
+				if (this.batchReleaseFuture == null) {
+					this.batchReleaseFuture =
+							Objects.requireNonNull(RabbitAmqpListenerContainer.this.taskScheduler)
+									.schedule(this::releaseInternal,
+											Instant.now().plus(RabbitAmqpListenerContainer.this.batchReceiveDuration));
+				}
+			}
+
+			void release() {
+				ScheduledFuture<?> currentBatchReleaseFuture = this.batchReleaseFuture;
+				if (currentBatchReleaseFuture != null) {
+					currentBatchReleaseFuture.cancel(true);
+					releaseInternal();
+				}
+			}
+
+			private void releaseInternal() {
+				if (this.batchReleaseFuture != null) {
+					this.batchReleaseFuture = null;
+					invokeBatchListener(this.batchContext, this.batch);
+				}
+			}
+
+		}
+
+	}
+
 }

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java

@@ -17,15 +17,22 @@
 package org.springframework.amqp.rabbitmq.client.listener;
 
 import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
 
 import com.rabbitmq.client.amqp.Consumer;
 import org.jspecify.annotations.Nullable;
 
+import org.springframework.amqp.core.AmqpAcknowledgment;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.rabbit.listener.adapter.InvocationResult;
 import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter;
 import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
 import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException;
 import org.springframework.amqp.rabbitmq.client.RabbitAmqpUtils;
+import org.springframework.messaging.support.GenericMessage;
 
 /**
  * A {@link MessagingMessageListenerAdapter} extension for the {@link RabbitAmqpMessageListener}.
@@ -45,15 +52,26 @@ import org.springframework.amqp.rabbitmq.client.RabbitAmqpUtils;
 public class RabbitAmqpMessageListenerAdapter extends MessagingMessageListenerAdapter
 		implements RabbitAmqpMessageListener {
 
-	public RabbitAmqpMessageListenerAdapter(@Nullable Object bean, @Nullable Method method, boolean returnExceptions,
-			@Nullable RabbitListenerErrorHandler errorHandler) {
+	private @Nullable Collection<MessagePostProcessor> afterReceivePostProcessors;
 
-		super(bean, method, returnExceptions, errorHandler);
+	public RabbitAmqpMessageListenerAdapter(@Nullable Object bean, @Nullable Method method, boolean returnExceptions,
+			@Nullable RabbitListenerErrorHandler errorHandler, boolean batch) {
+
+		super(bean, method, returnExceptions, errorHandler, batch);
+	}
+
+	public void setAfterReceivePostProcessors(Collection<MessagePostProcessor> afterReceivePostProcessors) {
+		this.afterReceivePostProcessors = new ArrayList<>(afterReceivePostProcessors);
 	}
 
 	@Override
 	public void onAmqpMessage(com.rabbitmq.client.amqp.Message amqpMessage, Consumer.@Nullable Context context) {
 		org.springframework.amqp.core.Message springMessage = RabbitAmqpUtils.fromAmqpMessage(amqpMessage, context);
+		if (this.afterReceivePostProcessors != null) {
+			for (MessagePostProcessor processor : this.afterReceivePostProcessors) {
+				springMessage = processor.postProcessMessage(springMessage);
+			}
+		}
 		try {
 			org.springframework.messaging.Message<?> messagingMessage = toMessagingMessage(springMessage);
 			InvocationResult result = getHandlerAdapter()
@@ -69,4 +87,46 @@ public class RabbitAmqpMessageListenerAdapter extends MessagingMessageListenerAd
 		}
 	}
 
+	@Override
+	public void onMessageBatch(List<Message> messages) {
+		AmqpAcknowledgment amqpAcknowledgment =
+				messages.stream()
+						.findAny()
+						.map((message) -> message.getMessageProperties().getAmqpAcknowledgment())
+						.orElse(null);
+
+		org.springframework.messaging.Message<?> converted;
+		if (this.messagingMessageConverter.isAmqpMessageList()) {
+			converted = new GenericMessage<>(messages);
+		}
+		else {
+			List<? extends org.springframework.messaging.Message<?>> messagingMessages =
+					messages.stream()
+							.map(this::toMessagingMessage)
+							.toList();
+
+			if (this.messagingMessageConverter.isMessageList()) {
+				converted = new GenericMessage<>(messagingMessages);
+			}
+			else {
+				List<Object> payloads = new ArrayList<>();
+				for (org.springframework.messaging.Message<?> message : messagingMessages) {
+					payloads.add(message.getPayload());
+				}
+				converted = new GenericMessage<>(payloads);
+			}
+		}
+		try {
+			InvocationResult result = getHandlerAdapter()
+					.invoke(converted, amqpAcknowledgment);
+			if (result.getReturnValue() != null) {
+				logger.warn("Replies are not currently supported with RabbitMQ AMQP 1.0 listeners");
+			}
+		}
+		catch (Exception ex) {
+			throw new ListenerExecutionFailedException("Failed to invoke listener", ex,
+					messages.toArray(new Message[0]));
+		}
+	}
+
 }

spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerTests.java

@@ -20,8 +20,10 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Random;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
 
 import com.rabbitmq.client.amqp.Connection;
 import com.rabbitmq.client.amqp.Consumer;
@@ -42,10 +44,13 @@ import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
 import org.springframework.amqp.rabbitmq.client.RabbitAmqpTestBase;
 import org.springframework.amqp.rabbitmq.client.config.RabbitAmqpListenerContainerFactory;
 import org.springframework.amqp.support.converter.MessageConversionException;
+import org.springframework.amqp.utils.test.TestUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.test.context.ContextConfiguration;
+import org.springframework.util.MultiValueMap;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -93,6 +98,53 @@ class RabbitAmqpListenerTests extends RabbitAmqpTestBase {
 		assertThat(this.template.receive("dlq1")).succeedsWithin(10, TimeUnit.SECONDS);
 	}
 
+	@Test
+	@SuppressWarnings("unchecked")
+	void verifyBatchConsumedAfterScheduledTimeout() {
+		List<String> testDataList =
+				List.of("batchData1", "batchData2", "batchData3", "batchData4", "batchData5");
+
+		for (String testData : testDataList) {
+			this.template.convertAndSend("q3", testData);
+		}
+
+		assertThat(this.config.batchReceived).succeedsWithin(10, TimeUnit.SECONDS)
+				.asInstanceOf(InstanceOfAssertFactories.LIST)
+				.hasSize(5)
+				.containsAll(testDataList);
+
+		assertThat(this.config.batchReceivedOnThread).startsWith("batch-consumer-scheduler-");
+
+		MessageListenerContainer testBatchListener =
+				this.rabbitListenerEndpointRegistry.getListenerContainer("testBatchListener");
+
+		MultiValueMap<String, Consumer> queueToConsumers =
+				TestUtils.getPropertyValue(testBatchListener, "queueToConsumers", MultiValueMap.class);
+		Consumer consumer = queueToConsumers.get("q3").get(0);
+
+		assertThat(consumer.unsettledMessageCount()).isEqualTo(0L);
+
+		this.config.batchReceived = new CompletableFuture<>();
+
+		testDataList =
+				IntStream.range(6, 16)
+						.boxed()
+						.map(Object::toString)
+						.map("batchData"::concat)
+						.toList();
+
+		for (String testData : testDataList) {
+			this.template.convertAndSend("q3", testData);
+		}
+
+		assertThat(this.config.batchReceived).succeedsWithin(10, TimeUnit.SECONDS)
+				.asInstanceOf(InstanceOfAssertFactories.LIST)
+				.hasSize(10)
+				.containsAll(testDataList);
+
+		assertThat(this.config.batchReceivedOnThread).startsWith("dispatching-rabbitmq-amqp-");
+	}
+
 	@Configuration
 	@EnableRabbit
 	static class Config {
@@ -122,6 +174,11 @@ class RabbitAmqpListenerTests extends RabbitAmqpTestBase {
 			return QueueBuilder.durable("q2").deadLetterExchange("dlx1").build();
 		}
 
+		@Bean
+		Queue q3() {
+			return new Queue("q3");
+		}
+
 		@Bean(RabbitListenerAnnotationBeanPostProcessor.DEFAULT_RABBIT_LISTENER_CONTAINER_FACTORY_BEAN_NAME)
 		RabbitAmqpListenerContainerFactory rabbitAmqpListenerContainerFactory(Connection connection) {
 			return new RabbitAmqpListenerContainerFactory(connection);
@@ -158,6 +215,38 @@ class RabbitAmqpListenerTests extends RabbitAmqpTestBase {
 			}
 		}
 
+		@Bean
+		ThreadPoolTaskScheduler taskScheduler() {
+			ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
+			threadPoolTaskScheduler.setPoolSize(2);
+			threadPoolTaskScheduler.setThreadNamePrefix("batch-consumer-scheduler-");
+			return threadPoolTaskScheduler;
+		}
+
+		@Bean
+		RabbitAmqpListenerContainerFactory batchRabbitAmqpListenerContainerFactory(Connection connection,
+				ThreadPoolTaskScheduler taskScheduler) {
+
+			RabbitAmqpListenerContainerFactory rabbitAmqpListenerContainerFactory =
+					new RabbitAmqpListenerContainerFactory(connection);
+			rabbitAmqpListenerContainerFactory.setTaskScheduler(taskScheduler);
+			rabbitAmqpListenerContainerFactory.setBatchSize(10);
+			rabbitAmqpListenerContainerFactory.setBatchReceiveTimeout(1000L);
+			return rabbitAmqpListenerContainerFactory;
+		}
+
+		CompletableFuture<List<String>> batchReceived = new CompletableFuture<>();
+
+		volatile String batchReceivedOnThread;
+
+		@RabbitListener(queues = "q3",
+				containerFactory = "batchRabbitAmqpListenerContainerFactory",
+				id = "testBatchListener")
+		void processBatchFromQ3(List<String> data) {
+			this.batchReceivedOnThread = Thread.currentThread().getName();
+			this.batchReceived.complete(data);
+		}
+
 	}
 
 }