spring-projectsGH-2601: Add a batchReceiveTimeout

Fixes spring-projects#2601

stop to waiting next message and execute listener when batchReceiveTimeout is timed out.

Author: laststem

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2022 the original author or authors.
+ * Copyright 2016-2024 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.
@@ -55,6 +55,7 @@
  * @author Gary Russell
  * @author Artem Bilan
  * @author Johno Crawford
+ * @author Jeonggi Kim
  *
  * @since 2.0
  *
@@ -166,6 +167,8 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean<AbstractMe
 
 	private Long receiveTimeout;
 
+	private Long batchReceiveTimeout;
+
 	private Integer batchSize;
 
 	private Integer declarationRetries;
@@ -389,6 +392,10 @@ public void setReceiveTimeout(long receiveTimeout) {
 		this.receiveTimeout = receiveTimeout;
 	}
 
+	public void setBatchReceiveTimeout(long batchReceiveTimeout) {
+		this.batchReceiveTimeout = batchReceiveTimeout;
+	}
+
 	/**
 	 * This property has several functions.
 	 * <p>
@@ -552,6 +559,7 @@ private AbstractMessageListenerContainer createContainer() {
 					.acceptIfNotNull(this.consecutiveActiveTrigger, container::setConsecutiveActiveTrigger)
 					.acceptIfNotNull(this.consecutiveIdleTrigger, container::setConsecutiveIdleTrigger)
 					.acceptIfNotNull(this.receiveTimeout, container::setReceiveTimeout)
+					.acceptIfNotNull(this.batchReceiveTimeout, container::setBatchReceiveTimeout)
 					.acceptIfNotNull(this.batchSize, container::setBatchSize)
 					.acceptIfNotNull(this.consumerBatchEnabled, container::setConsumerBatchEnabled)
 					.acceptIfNotNull(this.declarationRetries, container::setDeclarationRetries)

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2022 the original author or authors.
+ * Copyright 2014-2024 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.
@@ -32,6 +32,7 @@
  * @author Gary Russell
  * @author Artem Bilan
  * @author Dustin Schultz
+ * @author Jeonggi Kim
  *
  * @since 1.4
  */
@@ -54,6 +55,8 @@ public class SimpleRabbitListenerContainerFactory
 
 	private Long receiveTimeout;
 
+	private Long batchReceiveTimeout;
+
 	private Boolean consumerBatchEnabled;
 
 	/**
@@ -121,6 +124,14 @@ public void setReceiveTimeout(Long receiveTimeout) {
 		this.receiveTimeout = receiveTimeout;
 	}
 
+	/**
+	 * @param batchReceiveTimeout the timeout for gather batch messages.
+	 * @see SimpleMessageListenerContainer#setBatchReceiveTimeout
+	 */
+	public void setBatchReceiveTimeout(Long batchReceiveTimeout) {
+		this.batchReceiveTimeout = batchReceiveTimeout;
+	}
+
 	/**
 	 * Set to true to present a list of messages based on the {@link #setBatchSize(Integer)},
 	 * if the listener supports it. Starting with version 3.0, setting this to true will
@@ -163,7 +174,8 @@ protected void initializeContainer(SimpleMessageListenerContainer instance, Rabb
 			.acceptIfNotNull(this.stopConsumerMinInterval, instance::setStopConsumerMinInterval)
 			.acceptIfNotNull(this.consecutiveActiveTrigger, instance::setConsecutiveActiveTrigger)
 			.acceptIfNotNull(this.consecutiveIdleTrigger, instance::setConsecutiveIdleTrigger)
-			.acceptIfNotNull(this.receiveTimeout, instance::setReceiveTimeout);
+			.acceptIfNotNull(this.receiveTimeout, instance::setReceiveTimeout)
+			.acceptIfNotNull(this.batchReceiveTimeout, instance::setBatchReceiveTimeout);
 		if (Boolean.TRUE.equals(this.consumerBatchEnabled)) {
 			instance.setConsumerBatchEnabled(true);
 			/*

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2023 the original author or authors.
+ * Copyright 2002-2024 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.
@@ -80,6 +80,7 @@
  * @author Mat Jaggard
  * @author Yansong Ren
  * @author Tim Bourquin
+ * @author Jeonggi Kim
  *
  * @since 1.0
  */
@@ -97,6 +98,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 
 	private static final int DEFAULT_CONSECUTIVE_IDLE_TRIGGER = 10;
 
+	private static final long DEFAULT_BATCH_RECEIVE_TIMEOUT = 3000;
+
 	public static final long DEFAULT_RECEIVE_TIMEOUT = 1000;
 
 	private final AtomicLong lastNoMessageAlert = new AtomicLong();
@@ -121,6 +124,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 
 	private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
 
+	private long batchReceiveTimeout = DEFAULT_BATCH_RECEIVE_TIMEOUT;
+
 	private Set<BlockingQueueConsumer> consumers;
 
 	private Integer declarationRetries;
@@ -330,6 +335,16 @@ public void setReceiveTimeout(long receiveTimeout) {
 		this.receiveTimeout = receiveTimeout;
 	}
 
+	/**
+	 * The time (in milliseconds) that a waiting time to fill batch size. Default
+	 * 3000 (3 second).
+	 * @param batchReceiveTimeout the timeout
+	 */
+	public void setBatchReceiveTimeout(long batchReceiveTimeout) {
+		Assert.isTrue(batchReceiveTimeout > 0, "'batchReceiveTimeout' must be > 0");
+		this.batchReceiveTimeout = batchReceiveTimeout;
+	}
+
 	/**
 	 * This property has several functions.
 	 * <p>
@@ -996,8 +1011,14 @@ private boolean doReceiveAndExecute(BlockingQueueConsumer consumer) throws Excep
 
 		List<Message> messages = null;
 		long deliveryTag = 0;
+		long startTime = System.currentTimeMillis();
 
 		for (int i = 0; i < this.batchSize; i++) {
+			boolean batchTimedOut = (System.currentTimeMillis() - startTime) > this.batchReceiveTimeout;
+			if (batchTimedOut) {
+				logger.trace("Timed out for gather batch messages.");
+				break;
+			}
 
 			logger.trace("Waiting for message from consumer.");
 			Message message = consumer.nextMessage(this.receiveTimeout);

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

@@ -110,6 +110,7 @@
  * @author Mohammad Hewedy
  * @author Yansong Ren
  * @author Tim Bourquin
+ * @author Jeonggi Kim
  */
 public class SimpleMessageListenerContainerTests {
 
@@ -784,6 +785,59 @@ void testWithConsumerStartWhenNotActive() {
 		assertThat(start.getCount()).isEqualTo(0L);
 	}
 
+	@Test
+	public void testBatchReceiveTimedOut() throws Exception {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		Connection connection = mock(Connection.class);
+		Channel channel = mock(Channel.class);
+		given(connectionFactory.createConnection()).willReturn(connection);
+		given(connection.createChannel(false)).willReturn(channel);
+		final AtomicReference<Consumer> consumer = new AtomicReference<>();
+		willAnswer(invocation -> {
+			consumer.set(invocation.getArgument(6));
+			consumer.get().handleConsumeOk("1");
+			return "1";
+		}).given(channel)
+				.basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(), anyMap(),
+						any(Consumer.class));
+		final CountDownLatch latch = new CountDownLatch(2);
+		willAnswer(invocation -> {
+			latch.countDown();
+			return null;
+		}).given(channel).basicAck(anyLong(), anyBoolean());
+
+		final SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
+		container.setAfterReceivePostProcessors(msg -> null);
+		container.setQueueNames("foo");
+		MessageListener listener = mock(BatchMessageListener.class);
+		container.setMessageListener(listener);
+		container.setBatchSize(3);
+		container.setConsumerBatchEnabled(true);
+		container.setReceiveTimeout(10);
+		container.setBatchReceiveTimeout(20);
+		container.start();
+
+		BasicProperties props = new BasicProperties();
+		byte[] payload = "baz".getBytes();
+		Envelope envelope = new Envelope(1L, false, "foo", "bar");
+		consumer.get().handleDelivery("1", envelope, props, payload);
+		envelope = new Envelope(2L, false, "foo", "bar");
+		consumer.get().handleDelivery("1", envelope, props, payload);
+		// waiting for batch receive timed out
+		Thread.sleep(20);
+		envelope = new Envelope(3L, false, "foo", "bar");
+		consumer.get().handleDelivery("1", envelope, props, payload);
+		assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+		verify(channel, never()).basicAck(eq(1), anyBoolean());
+		verify(channel).basicAck(2, true);
+		verify(channel, never()).basicAck(eq(2), anyBoolean());
+		verify(channel).basicAck(3, true);
+		container.stop();
+		verify(listener).containerAckMode(AcknowledgeMode.AUTO);
+		verify(listener).isAsyncReplies();
+		verifyNoMoreInteractions(listener);
+	}
+
 	private Answer<Object> messageToConsumer(final Channel mockChannel, final SimpleMessageListenerContainer container,
 			final boolean cancel, final CountDownLatch latch) {
 		return invocation -> {