GH-3152: Fix for nested Scatter Gather

Fixes #3152

The upstream `gatherResultChannel` header has been missed when we produced a reply from nested scatter-gather

Added Test Case for Nested Scatter Gather test

Simplified the the test cases and added author in changed cases

Corrected codestyle issue in Travis CI

Removed additional OriginalReplyChannel and originalErrorChannel in Headers.
Added additional not to be executed line of code in test case.

Restored OriginalErrorChannel Header and removed error handling related fixes

* Clean up code style and improve readability

**Cherry-pick to 5.1.x & master**

Author: sirimamilla

spring-integration-core/src/main/java/org/springframework/integration/scattergather/ScatterGatherHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2019 the original author or authors.
+ * Copyright 2014-2020 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.
@@ -48,15 +48,14 @@
  *
  * @author Artem Bilan
  * @author Abdul Zaheer
+ * @author Jayadev Sirimamilla
  *
  * @since 4.1
  */
 public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler implements Lifecycle {
 
 	private static final String GATHER_RESULT_CHANNEL = "gatherResultChannel";
 
-	private static final String ORIGINAL_REPLY_CHANNEL = "originalReplyChannel";
-
 	private static final String ORIGINAL_ERROR_CHANNEL = "originalErrorChannel";
 
 	private final MessageChannel scatterChannel;
@@ -167,9 +166,7 @@ private Message<?> enhanceScatterReplyMessage(Message<?> message) {
 		MessageHeaders headers = message.getHeaders();
 		return getMessageBuilderFactory()
 				.fromMessage(message)
-				.setHeader(MessageHeaders.REPLY_CHANNEL, headers.get(ORIGINAL_REPLY_CHANNEL))
 				.setHeader(MessageHeaders.ERROR_CHANNEL, headers.get(ORIGINAL_ERROR_CHANNEL))
-				.removeHeaders(ORIGINAL_REPLY_CHANNEL, ORIGINAL_ERROR_CHANNEL)
 				.build();
 	}
 
@@ -182,15 +179,22 @@ protected Object handleRequestMessage(Message<?> requestMessage) {
 				getMessageBuilderFactory()
 						.fromMessage(requestMessage)
 						.setHeader(GATHER_RESULT_CHANNEL, gatherResultChannel)
-						.setHeader(ORIGINAL_REPLY_CHANNEL, requestMessageHeaders.getReplyChannel())
 						.setHeader(ORIGINAL_ERROR_CHANNEL, requestMessageHeaders.getErrorChannel())
 						.setReplyChannel(this.gatherChannel)
 						.setErrorChannelName(this.errorChannelName)
 						.build();
 
 		this.messagingTemplate.send(this.scatterChannel, scatterMessage);
 
-		return gatherResultChannel.receive(this.gatherTimeout);
+		Message<?> gatherResult = gatherResultChannel.receive(this.gatherTimeout);
+		if (gatherResult != null) {
+			return getMessageBuilderFactory()
+					.fromMessage(gatherResult)
+					.removeHeaders(GATHER_RESULT_CHANNEL, ORIGINAL_ERROR_CHANNEL,
+							MessageHeaders.REPLY_CHANNEL, MessageHeaders.ERROR_CHANNEL);
+		}
+
+		return null;
 	}
 
 	@Override

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -42,9 +42,11 @@
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.config.EnableMessageHistory;
 import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlowDefinition;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.MessageChannels;
 import org.springframework.integration.expression.FunctionExpression;
+import org.springframework.integration.store.MessageGroup;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -62,6 +64,7 @@
 /**
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Jayadev Sirimamilla
  *
  * @since 5.0
  */
@@ -531,7 +534,6 @@ public void testExceptionTypeRouteFlow() {
 	private MessageChannel nestedScatterGatherFlowInput;
 
 	@Test
-	@SuppressWarnings("unchecked")
 	public void testNestedScatterGather() {
 		QueueChannel replyChannel = new QueueChannel();
 		Message<String> request = MessageBuilder.withPayload("this is a test")
@@ -578,7 +580,7 @@ public void testScatterGatherWithExecutorChannelSubFlow() {
 		assertThat(receive).isNotNull();
 		Object payload = receive.getPayload();
 		assertThat(payload).isInstanceOf(List.class);
-		assertThat(((List) payload).get(1)).isInstanceOf(RuntimeException.class);
+		assertThat(((List<?>) payload).get(1)).isInstanceOf(RuntimeException.class);
 	}
 
 	@Autowired
@@ -592,6 +594,25 @@ public void propagateErrorFromGatherer() {
 				.withMessage("intentional");
 	}
 
+	@Autowired
+	@Qualifier("scatterGatherInSubFlow.input")
+	MessageChannel scatterGatherInSubFlowChannel;
+
+
+	@Test
+	public void testNestedScatterGatherSuccess() {
+		PollableChannel replyChannel = new QueueChannel();
+		this.scatterGatherInSubFlowChannel.send(
+				org.springframework.integration.support.MessageBuilder.withPayload("baz")
+						.setReplyChannel(replyChannel)
+						.build());
+
+		Message<?> receive = replyChannel.receive(10000);
+		assertThat(receive).isNotNull();
+		assertThat(receive.getPayload()).isEqualTo("baz");
+
+	}
+
 	@Configuration
 	@EnableIntegration
 	@EnableMessageHistory({ "recipientListOrder*", "recipient1*", "recipient2*" })
@@ -896,9 +917,21 @@ public IntegrationFlow propagateErrorFromGatherer(TaskExecutor taskExecutor) {
 										throw new RuntimeException("intentional");
 									}),
 							sg -> sg.gatherTimeout(100))
+					.transform(m -> "This should not be executed, results must have been propagated to Error Channel")
 					.get();
 		}
 
+		@Bean
+		public IntegrationFlow scatterGatherInSubFlow() {
+			return flow -> flow.scatterGather(s -> s.applySequence(true)
+							.recipientFlow(inflow -> inflow
+									.scatterGather(s1 -> s1.applySequence(true)
+													.recipientFlow(IntegrationFlowDefinition::bridge),
+											g -> g.outputProcessor(MessageGroup::getOne)
+									)),
+					g -> g.outputProcessor(MessageGroup::getOne));
+		}
+
 	}
 
 	private static class RoutingTestBean {