INT-4542: CorrMH: Use popSequenceDetails properly

JIRA: https://jira.spring.io/browse/INT-4542

Perform a `MessageBuilder.popSequenceDetails()` for those
`MessageGroupProcessor` results which are not `Message`
or `Collection<Message>`

**Cherry-pick to 5.0.x**

Author: artembilan

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

@@ -51,6 +51,7 @@
 import org.springframework.integration.store.MessageStore;
 import org.springframework.integration.store.SimpleMessageGroup;
 import org.springframework.integration.store.SimpleMessageStore;
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.support.locks.DefaultLockRegistry;
 import org.springframework.integration.support.locks.LockRegistry;
 import org.springframework.integration.util.UUIDConverter;
@@ -653,7 +654,8 @@ protected void forceComplete(MessageGroup group) {
 						 * groups. A longer timeout for empty groups can be enabled by
 						 * setting minimumTimeoutForEmptyGroups.
 						 */
-						removeGroup = lastModifiedNow <= (System.currentTimeMillis() - this.minimumTimeoutForEmptyGroups);
+						removeGroup = lastModifiedNow <= (System
+								.currentTimeMillis() - this.minimumTimeoutForEmptyGroups);
 						if (removeGroup && this.logger.isDebugEnabled()) {
 							this.logger.debug("Removing empty group: " + correlationKey);
 						}
@@ -753,10 +755,24 @@ protected Collection<Message<?>> completeGroup(Message<?> message, Object correl
 		Object result = this.outputProcessor.processMessageGroup(group);
 		Collection<Message<?>> partialSequence = null;
 		if (result instanceof Collection<?>) {
-			this.verifyResultCollectionConsistsOfMessages((Collection<?>) result);
+			verifyResultCollectionConsistsOfMessages((Collection<?>) result);
 			partialSequence = (Collection<Message<?>>) result;
 		}
-		this.sendOutputs(result, message);
+
+		if (partialSequence == null && !(result instanceof Message<?>)) {
+			AbstractIntegrationMessageBuilder<?> messageBuilder;
+			if (result instanceof AbstractIntegrationMessageBuilder<?>) {
+				messageBuilder = (AbstractIntegrationMessageBuilder<?>) result;
+			}
+			else {
+				messageBuilder = getMessageBuilderFactory()
+						.withPayload(result)
+						.copyHeaders(message.getHeaders());
+			}
+			result = messageBuilder.popSequenceDetails();
+		}
+
+		sendOutputs(result, message);
 		return partialSequence;
 	}
 
@@ -772,10 +788,8 @@ protected Long obtainGroupTimeout(MessageGroup group) {
 	}
 
 	@Override
-	public void destroy() throws Exception {
-		for (ScheduledFuture<?> future : this.expireGroupScheduledFutures.values()) {
-			future.cancel(true);
-		}
+	public void destroy() {
+		this.expireGroupScheduledFutures.values().forEach(future -> future.cancel(true));
 	}
 
 	@Override

spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java

@@ -29,6 +29,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -531,6 +532,25 @@ public void testExceptionTypeRouteFlow() {
 		assertNull(this.messageHandlingExceptionChannel.receive(0));
 	}
 
+	@Autowired
+	@Qualifier("nestedScatterGatherFlow.input")
+	private MessageChannel nestedScatterGatherFlowInput;
+
+	@Test
+	public void testNestedScatterGather() {
+		QueueChannel replyChannel = new QueueChannel();
+		Message<String> request = MessageBuilder.withPayload("this is a test")
+				.setReplyChannel(replyChannel)
+				.build();
+		this.nestedScatterGatherFlowInput.send(request);
+		Message<?> bestQuoteMessage = replyChannel.receive(10000);
+		assertNotNull(bestQuoteMessage);
+		Object payload = bestQuoteMessage.getPayload();
+		assertThat(payload, instanceOf(String.class));
+//		assertThat(((List<?>) payload).size(), greaterThanOrEqualTo(1));
+	}
+
+
 	@Configuration
 	@EnableIntegration
 	@EnableMessageHistory({ "recipientListOrder*", "recipient1*", "recipient2*" })
@@ -773,6 +793,29 @@ public IntegrationFlow scatterGatherFlow() {
 									.gatherTimeout(10_000));
 		}
 
+		@Bean
+		public IntegrationFlow nestedScatterGatherFlow() {
+			return f -> f
+					.split(s -> s.delimiters(" "))
+					.scatterGather(
+							scatterer -> scatterer
+									.recipientFlow(f1 -> f1.handle((p, h) -> p + " - flow 1"))
+									.recipientFlow(f2 -> f2.handle((p, h) -> p + " - flow 2"))
+									.applySequence(true),
+							gatherer -> gatherer
+									.outputProcessor(mg -> mg
+											.getMessages()
+											.stream()
+											.map(m -> m.getPayload().toString())
+											.collect(Collectors.joining(", "))),
+							scatterGather -> scatterGather.gatherTimeout(10_000))
+					.aggregate()
+					.<List<String>, String>transform(source ->
+							source.stream()
+									.map(s -> "- " + s)
+									.collect(Collectors.joining("\n")));
+		}
+
 	}
 
 	private static class RoutingTestBean {

src/reference/asciidoc/aggregator.adoc

@@ -103,6 +103,10 @@ This method is invoked for aggregating messages as follows:
 
 NOTE: In the interest of code simplicity and promoting best practices such as low coupling, testability, and others, the preferred way of implementing the aggregation logic is through a POJO and using the XML or annotation support for configuring it in the application.
 
+IMPORTANT: After processing message group, an `AbstractCorrelatingMessageHandler` performs a `MessageBuilder.popSequenceDetails()` message headers modification for the proper splitter-aggregator scenario with several nested levels.
+It is done only if the message group release result is not a message or collection of messages.
+In that case a target `MessageGroupProcessor` is responsible for the `MessageBuilder.popSequenceDetails()` call during those messages building.
+
 [[agg-message-collection]]
 IMPORTANT: The `SimpleMessageGroup.getMessages()` method returns an `unmodifiableCollection`.
 Therefore, if your aggregating POJO method has a `Collection<Message>` parameter, the argument passed in is exactly that `Collection` instance and, when you use a `SimpleMessageStore` for the aggregator, that original `Collection<Message>` is cleared after releasing the group.