GH-3152: Fix for nested Scatter Gather

Fixes https://github.com/spring-projects/spring-integration/issues/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**

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

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,6 +48,7 @@ import org.springframework.util.ClassUtils;
  *
  * @author Artem Bilan
  * @author Abdul Zaheer
+ * @author Jayadev Sirimamilla
  *
  * @since 4.1
  */
@@ -55,8 +56,6 @@ public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler i
 
 	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 @@ public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler i
 		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,7 +179,6 @@ public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler i
 				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)
@@ -190,7 +186,15 @@ public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler i
 
 		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.
@@ -49,9 +49,11 @@ import org.springframework.integration.channel.QueueChannel;
 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;
@@ -69,6 +71,7 @@ import org.springframework.test.context.junit4.SpringRunner;
 /**
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Jayadev Sirimamilla
  *
  * @since 5.0
  */
@@ -542,7 +545,6 @@ public class RouterTests {
 	private MessageChannel nestedScatterGatherFlowInput;
 
 	@Test
-	@SuppressWarnings("unchecked")
 	public void testNestedScatterGather() {
 		QueueChannel replyChannel = new QueueChannel();
 		Message<String> request = MessageBuilder.withPayload("this is a test")
@@ -602,6 +604,25 @@ public class RouterTests {
 				.hasMessage("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);
+		assertNotNull(receive);
+		assertEquals("baz", receive.getPayload());
+
+	}
+
 	@Configuration
 	@EnableIntegration
 	@EnableMessageHistory({ "recipientListOrder*", "recipient1*", "recipient2*" })
@@ -906,9 +927,21 @@ public class RouterTests {
 										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 {