GH-9754: Add discardIndividuallyOnExpiry to aggregator

Fixes: https://github.com/spring-projects/spring-integration/issues/9754

Right now a correlation handler can discard messages in the expired group one by one.
In some scenarios it would be useful to have single message in discard for the whole group.

* Expose `discardIndividuallyOnExpiry` for the `AbstractCorrelatingMessageHandler`,
and `AggregatorFactoryBean`, and respective `CorrelationHandlerSpec` for DSL.
This new option takes action only if a `discardChannel` is provided,
and `sendPartialResultOnExpiry` is not set to `true`.
When `discardIndividuallyOnExpiry` is false, the messages in the expired group are packed
into a list for payload of a discarding single message.
* Test and document the new feature

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java

@@ -18,6 +18,7 @@ package org.springframework.integration.aggregator;
 
 import java.time.Duration;
 import java.time.Instant;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
@@ -63,6 +64,7 @@ import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.core.DestinationResolutionException;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.ObjectUtils;
@@ -80,7 +82,7 @@ import org.springframework.util.ObjectUtils;
  * {@link ReleaseStrategy}, and {@link MessageGroupProcessor} implementations as
  * you require.
  * <p>
- * By default the {@link CorrelationStrategy} will be a
+ * By default, the {@link CorrelationStrategy} will be a
  * {@link HeaderAttributeCorrelationStrategy} and the {@link ReleaseStrategy} will be a
  * {@link SequenceSizeReleaseStrategy}.
  * <p>
@@ -129,6 +131,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 
 	private boolean sendPartialResultOnExpiry;
 
+	private boolean discardIndividuallyOnExpiry = true;
+
 	private boolean sequenceAware;
 
 	private LockRegistry lockRegistry = new DefaultLockRegistry();
@@ -262,6 +266,18 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 		this.sendPartialResultOnExpiry = sendPartialResultOnExpiry;
 	}
 
+	/**
+	 * Set to {@code false} to send to discard channel a whole expired group as a single message.
+	 * This option makes sense only if {@link #sendPartialResultOnExpiry} is set to {@code false} (default).
+	 * And also if {@link #discardChannel} is injected.
+	 * @param discardIndividuallyOnExpiry false to discard the whole group as one message.
+	 * @since 6.5
+	 * @see #sendPartialResultOnExpiry
+	 */
+	public void setDiscardIndividuallyOnExpiry(boolean discardIndividuallyOnExpiry) {
+		this.discardIndividuallyOnExpiry = discardIndividuallyOnExpiry;
+	}
+
 	/**
 	 * By default, when a MessageGroupStoreReaper is configured to expire partial
 	 * groups, empty groups are also removed. Empty groups exist after a group
@@ -876,8 +892,17 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 			if (this.releaseLockBeforeSend) {
 				lock.unlock();
 			}
-			group.getMessages()
-					.forEach(this::discardMessage);
+			MessageChannel messageChannel = getDiscardChannel();
+			if (messageChannel != null) {
+				if (this.discardIndividuallyOnExpiry) {
+					group.getMessages()
+							.forEach(this::discardMessage);
+				}
+				else {
+					List<Message<?>> messagesInGroupToDiscard = new ArrayList<>(group.getMessages());
+					discardMessage(new GenericMessage<>(messagesInGroupToDiscard));
+				}
+			}
 		}
 		if (this.applicationEventPublisher != null) {
 			this.applicationEventPublisher.publishEvent(

spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2024 the original author or authors.
+ * Copyright 2015-2025 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 class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe
 
 	private Boolean sendPartialResultOnExpiry;
 
+	private Boolean discardIndividuallyOnExpiry;
+
 	private Long minimumTimeoutForEmptyGroups;
 
 	private Boolean expireGroupsUponTimeout;
@@ -195,6 +197,16 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe
 		this.groupConditionSupplier = groupConditionSupplier;
 	}
 
+	/**
+	 * Set to {@code false} to send to discard channel a whole expired group as a single message.
+	 * @param discardIndividuallyOnExpiry  false to discard the whole group as one message.
+	 * @since 6.5
+	 * @see org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler#setDiscardIndividuallyOnExpiry(boolean)
+	 */
+	public void setDiscardIndividuallyOnExpiry(Boolean discardIndividuallyOnExpiry) {
+		this.discardIndividuallyOnExpiry = discardIndividuallyOnExpiry;
+	}
+
 	@Override
 	protected AggregatingMessageHandler createHandler() {
 		MessageGroupProcessor outputProcessor;
@@ -242,7 +254,8 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe
 				.acceptIfNotNull(this.expireDuration,
 						(duration) -> aggregator.setExpireDuration(Duration.ofMillis(duration)))
 				.acceptIfNotNull(this.groupConditionSupplier, aggregator::setGroupConditionSupplier)
-				.acceptIfNotNull(this.expireTimeout, aggregator::setExpireTimeout);
+				.acceptIfNotNull(this.expireTimeout, aggregator::setExpireTimeout)
+				.acceptIfNotNull(this.discardIndividuallyOnExpiry, aggregator::setDiscardIndividuallyOnExpiry);
 
 		return aggregator;
 	}

spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2022 the original author or authors.
+ * Copyright 2016-2025 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.
@@ -85,6 +85,19 @@ public abstract class CorrelationHandlerSpec<S extends CorrelationHandlerSpec<S,
 		return _this();
 	}
 
+	/**
+	 * Set to {@code false} to send to discard channel a whole expired group as a single message.
+	 * This option makes sense only if {@link #sendPartialResultOnExpiry(boolean)} is set to {@code false} (default).
+	 * And also if {@link #discardChannel(MessageChannel)} is injected.
+	 * @param discardIndividuallyOnExpiry false to discard whole expired group as a single message.
+	 * @return the handler spec.
+	 * @since 6.5
+	 */
+	public S discardIndividuallyOnExpiry(boolean discardIndividuallyOnExpiry) {
+		this.handler.setDiscardIndividuallyOnExpiry(discardIndividuallyOnExpiry);
+		return _this();
+	}
+
 	/**
 	 * @param minimumTimeoutForEmptyGroups the minimumTimeoutForEmptyGroups
 	 * @return the handler spec.

spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java

@@ -27,6 +27,7 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
+import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
@@ -531,4 +532,31 @@ public class AbstractCorrelatingMessageHandlerTests {
 		taskScheduler.destroy();
 	}
 
+	@Test
+	public void expiredGroupIsDiscardedAsOneMessage() throws InterruptedException {
+		AggregatingMessageHandler handler = new AggregatingMessageHandler(group -> group);
+		handler.setReleaseStrategy(group -> false);
+		QueueChannel discardChannel = new QueueChannel();
+		handler.setDiscardChannel(discardChannel);
+		handler.setExpireTimeout(1);
+		handler.setDiscardIndividuallyOnExpiry(false);
+
+		Message<String> message1 = MessageBuilder.withPayload("test1").setCorrelationId("test").build();
+		Message<String> message2 = MessageBuilder.withPayload("test2").setCorrelationId("test").build();
+
+		handler.handleMessageInternal(message1);
+		handler.handleMessageInternal(message2);
+
+		// Slight delay to let the group be treated as expired.
+		Thread.sleep(100);
+
+		handler.purgeOrphanedGroups();
+
+		Message<?> receive = discardChannel.receive(10000);
+		assertThat(receive)
+				.extracting(Message::getPayload)
+				.asInstanceOf(InstanceOfAssertFactories.LIST)
+				.containsOnly(message1, message2);
+	}
+
 }