GH-9754: Add discardIndividuallyOnExpiry to aggregator

Fixes: #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

Author: artembilan

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

@@ -18,6 +18,7 @@
 
 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.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 @@
  * {@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 void setSendPartialResultOnExpiry(boolean sendPartialResultOnExpiry) {
 		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 @@ protected void expireGroup(Object correlationKey, MessageGroup group, Lock lock)
 			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(

Diff for: 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 void setGroupConditionSupplier(BiFunction<Message<?>, String, String> gro
 		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 @@ protected AggregatingMessageHandler createHandler() {
 				.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;
 	}

Diff for: 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 S sendPartialResultOnExpiry(boolean sendPartialResultOnExpiry) {
 		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.

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

@@ -27,6 +27,7 @@
 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 void testPurgeOrphanedGroupsScheduled() {
 		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);
+	}
+
 }

Diff for: src/reference/antora/modules/ROOT/pages/aggregator.adoc

@@ -213,6 +213,10 @@ Any new messages for this group are sent to the discard channel (if defined).
 Setting `expire-groups-upon-completion` to `true` (the default is `false`) removes the entire group, and any new messages (with the same correlation ID as the removed group) form a new group.
 You can release partial sequences by using a `MessageGroupStoreReaper` together with `send-partial-result-on-expiry` being set to `true`.
 
+Starting with version 6.5, the correlation handler can also be configured with a `discardIndividuallyOnExpiry` option to discard the whole group as a single message.
+Essentially, the payload of this message is a list of messages from the expired group.
+Works only if `sendPartialResultOnExpiry` is set to `false` (default) and `dicardChannel` is provided.
+
 IMPORTANT: To facilitate discarding of late-arriving messages, the aggregator must maintain state about the group after it has been released.
 This can eventually cause out-of-memory conditions.
 To avoid such situations, you should consider configuring a `MessageGroupStoreReaper` to remove the group metadata.
@@ -519,7 +523,7 @@ Empty groups can be removed later by using a `MessageGroupStoreReaper` in combin
 `expire-groups-upon-completion` relates to "`normal`" completion when the `ReleaseStrategy` releases the group.
 This defaults to `false`.
 
-If a group is not completed normally but is released or discarded because of a timeout, the group is normally expired.
+If a group is not complete normally but is released or discarded because of a timeout, the group is normally expired.
 Since version 4.1, you can control this behavior by using `expire-groups-upon-timeout`.
 It defaults to `true` for backwards compatibility.
 
@@ -531,12 +535,12 @@ Timed-out groups are either discarded or a partial release occurs (based on `sen
 Since version 5.0, empty groups are also scheduled for removal after `empty-group-min-timeout`.
 If `expireGroupsUponCompletion == false` and `minimumTimeoutForEmptyGroups > 0`, the task to remove the group is scheduled when normal or partial sequences release happens.
 
-Starting with version 5.4, the aggregator (and resequencer) can be configured to expire orphaned groups (groups in a persistent message store that might not otherwise be released).
+Starting with version 5.4, the aggregator (and resequencer) can be configured to expire orphaned groups (those in a persistent message store that might not otherwise be released).
 The `expireTimeout` (if greater than `0`) indicates that groups older than this value in the store should be purged.
 The `purgeOrphanedGroups()` method is called on start up and, together with the provided `expireDuration`, periodically within a scheduled task.
 This method is also can be called externally at any time.
 The expiration logic is fully delegated to the `forceComplete(MessageGroup)` functionality according to the provided expiration options mentioned above.
-Such a periodic purge functionality is useful when a message store is needed to be cleaned up from those old groups which are not going to be released any more with regular message arrival logic.
+Such a periodic purge functionality is useful when a message store is needed to be cleaned up from those old groups which are not going to be released anymore with regular message arrival logic.
 In most cases this happens after an application restart, when using a persistent message group store.
 The functionality is similar to the `MessageGroupStoreReaper` with a scheduled task, but provides a convenient way to deal with old groups within specific components, when using group timeout instead of a reaper.
 The `MessageGroupStore` must be provided exclusively for the current correlation endpoint.
@@ -695,7 +699,7 @@ Otherwise, it is discarded.
 There is a difference between `groupTimeout` behavior and `MessageGroupStoreReaper` (see xref:aggregator.adoc#aggregator-xml[Configuring an Aggregator with XML]).
 The reaper initiates forced completion for all `MessageGroup` s in the `MessageGroupStore` periodically.
 The `groupTimeout` does it for each `MessageGroup` individually if a new message does not arrive during the `groupTimeout`.
-Also, the reaper can be used to remove empty groups (empty groups are retained in order to discard late messages if `expire-groups-upon-completion` is false).
+Also, the reaper can be used to remove empty groups (those retained in order to discard late messages if `expire-groups-upon-completion` is false).
 
 Starting with version 5.5, the `groupTimeoutExpression` can be evaluated to a `java.util.Date` instance.
 This can be useful in cases like determining a scheduled task moment based on the group creation time (`MessageGroup.getTimestamp()`) instead of a current message arrival as it is calculated when `groupTimeoutExpression` is evaluated to `long`:

Diff for: src/reference/antora/modules/ROOT/pages/whats-new.adoc

@@ -27,6 +27,12 @@ The `AbstractCorrelatingMessageHandler` does not throw an `IllegalArgumentExcept
 Instead, such a collection is wrapped into a single reply message.
 See xref:aggregator.adoc[Aggregator] for more information.
 
+[[x6.4-correlation-changes]]
+== The `discardIndividuallyOnExpiry` Option For Correlation Handlers
+
+The aggregator and resequencer can now discard the whole expired group as a single message via setting `discardIndividuallyOnExpiry` to `false`.
+See xref:aggregator.adoc#releasestrategy[ReleaseStrategy] for more information.
+
 [[x6.4-message-store-with-locks]]
 == The `LockRegistry` in the `MessageStore`