GH-9706: Allow collection of payloads as an aggregator result

Fixes: #9706

Previously, if a `MessageGroupProcessor` returns a collection of payloads,
the `AbstractCorrelatingMessageHandler` has failed with the `IllegalArgumentException`
stating that only collection of messages is possible.
From now on such a restriction is eliminated and returned collection of payloads
is emitted as a single reply message from the aggregator.

* Add `AbstractCorrelatingMessageHandler.isResultCollectionOfMessages()`
to return `true` only if result is a collection of messages, treating them as a "partial sequence".
* Deprecate `AbstractCorrelatingMessageHandler.verifyResultCollectionConsistsOfMessages()`
since it is out of use now.

Author: artembilan

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2024 the original author or authors.
+ * Copyright 2002-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.
@@ -904,8 +904,7 @@ protected Collection<Message<?>> completeGroup(Message<?> message, Object correl
 			this.logger.debug(() -> "Completing group with correlationKey [" + correlationKey + "]");
 
 			result = this.outputProcessor.processMessageGroup(group);
-			if (result instanceof Collection<?>) {
-				verifyResultCollectionConsistsOfMessages((Collection<?>) result);
+			if (isResultCollectionOfMessages(result)) {
 				partialSequence = (Collection<Message<?>>) result;
 			}
 
@@ -944,12 +943,26 @@ private static boolean compareSequences(Message<?> msg1, Message<?> msg2) {
 
 	}
 
+	/**
+	 * Probably the method is {@code protected} by mistake.
+	 * @param elements the group processor result.
+	 * @deprecated without replacement - out of use from now on.
+	 */
+	@Deprecated(since = "6.5", forRemoval = true)
 	protected void verifyResultCollectionConsistsOfMessages(Collection<?> elements) {
 		Class<?> commonElementType = CollectionUtils.findCommonElementType(elements);
 		Assert.isAssignable(Message.class, commonElementType, () ->
 				"The expected collection of Messages contains non-Message element: " + commonElementType);
 	}
 
+	private static boolean isResultCollectionOfMessages(Object result) {
+		if (result instanceof Collection<?> resultCollection) {
+			Class<?> commonElementType = CollectionUtils.findCommonElementType(resultCollection);
+			return commonElementType != null && Message.class.isAssignableFrom(commonElementType);
+		}
+		return false;
+	}
+
 	protected Object obtainGroupTimeout(MessageGroup group) {
 		if (this.groupTimeoutExpression != null) {
 			Object timeout = this.groupTimeoutExpression.getValue(this.evaluationContext, group);

Diff for: spring-integration-core/src/test/java/org/springframework/integration/dsl/extensions/IntegrationFlowExtensionTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2022 the original author or authors.
+ * Copyright 2020-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.
@@ -18,8 +18,8 @@
 
 import java.util.Arrays;
 import java.util.function.Consumer;
-import java.util.stream.Collectors;
 
+import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
@@ -66,7 +66,8 @@ public void testCustomFlowDefinition() {
 		assertThat(replyMessage)
 				.isNotNull()
 				.extracting(Message::getPayload)
-				.isEqualTo("ONE, TWO, THREE");
+				.asInstanceOf(InstanceOfAssertFactories.LIST)
+				.containsOnly("ONE", "TWO", "THREE");
 	}
 
 	@Configuration
@@ -109,8 +110,7 @@ public static class CustomAggregatorSpec extends AggregatorSpec {
 					group.getMessages()
 							.stream()
 							.map(Message::getPayload)
-							.map(String.class::cast)
-							.collect(Collectors.joining(", ")));
+							.toList());
 		}
 
 	}

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

@@ -144,6 +144,10 @@ Starting with version 6.0, the splitting behaviour, described above, works only
 Otherwise, with any other `MessageGroupProcessor` implementation that returns a `Collection<Message>`, only a single reply message is emitted with the whole collection of messages as its payload.
 Such logic is dictated by the canonical purpose of an aggregator - collect request messages by some key and produce a single grouped message.
 
+Prior to version 6.5, if a `MessageGroupProcessor` (usually lambda from DSL) returns a collection of payloads, the `AbstractCorrelatingMessageHandler` has failed with the `IllegalArgumentException` stating that only collection of messages is possible.
+From now on such a restriction is eliminated and returned collection of payloads is emitted as a single reply message from the aggregator with just headers from the last request message.
+If headers aggregation is required alongside with a collection of payloads, an `AbstractAggregatingMessageGroupProcessor` implementations are recommended to be used instead of plain `MessageGroupProcessor` functional interface.
+
 [[releasestrategy]]
 === `ReleaseStrategy`
 

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

@@ -21,3 +21,8 @@ The deprecated previously usage of `org.springframework.util.concurrent.Listenab
 The previously deprecated SpEL-based Control Bus components have been removed in favor of functionality around `ControlBusCommandRegistry`.
 The `<control-bus use-registry="">` attribute is deprecated now without replacement since only `ControlBusCommandRegistry` functionality is available.
 The Java DSL `controlBusOnRegistry()` operator is deprecated in favor of restored `controlBus()` which is fully based now on the `ControlBusCommandRegistry`.
+See xref:control-bus.adoc[Control Bus] for more information.
+
+The `AbstractCorrelatingMessageHandler` does not throw an `IllegalArgumentException` for the collection of payloads as a result of the `MessageGroupProcessor`.
+Instead, such a collection is wrapped into a single reply message.
+See xref:aggregator.adoc[Aggregator] for more information.