Improve MessageChannelPartitionHandler::receiveReplies

1. Message payload not always be `Set`, if you return payload directly, you will see tests failed:
```
[main] ERROR org.springframework.batch.core.step.AbstractStep - Encountered an error executing step step1-manager in job job1
java.lang.ClassCastException: class java.util.ArrayList cannot be cast to class java.util.Set (java.util.ArrayList and java.util.Set are in module java.base of loader 'bootstrap')
	at org.springframework.batch.integration.partition.MessageChannelPartitionHandler.receiveReplies(MessageChannelPartitionHandler.java:298)
	at org.springframework.batch.integration.partition.MessageChannelPartitionHandler.doHandle(MessageChannelPartitionHandler.java:244)
	at org.springframework.batch.core.partition.support.AbstractPartitionHandler.handle(AbstractPartitionHandler.java:60)
	at org.springframework.batch.core.partition.support.PartitionStep.doExecute(PartitionStep.java:102)
	at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:230)
	at org.springframework.batch.core.job.SimpleStepHandler.handleStep(SimpleStepHandler.java:153)
	at org.springframework.batch.core.job.flow.JobFlowExecutor.executeStep(JobFlowExecutor.java:68)
	at org.springframework.batch.core.job.flow.support.state.StepState.handle(StepState.java:68)
	at org.springframework.batch.core.job.flow.support.SimpleFlow.resume(SimpleFlow.java:165)
	at org.springframework.batch.core.job.flow.support.SimpleFlow.start(SimpleFlow.java:140)
	at org.springframework.batch.core.job.flow.FlowJob.doExecute(FlowJob.java:132)
	at org.springframework.batch.core.job.AbstractJob.execute(AbstractJob.java:307)
	at org.springframework.batch.core.launch.support.TaskExecutorJobLauncher$1.run(TaskExecutorJobLauncher.java:155)
	at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
	at org.springframework.batch.core.launch.support.TaskExecutorJobLauncher.run(TaskExecutorJobLauncher.java:146)
	at org.springframework.batch.integration.partition.VanillaIntegrationTests.testLaunchJob(VanillaIntegrationTests.java:58)
```

2. No need to create a new `HashSet` if payload is `Set` since no modifications applied to it.

Signed-off-by: Yanming Zhou <zhouyanming@gmail.com>
(cherry picked from commit 0d3832f)

Author: quaff

spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandler.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.batch.integration.partition;
 
+import java.util.Collection;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
@@ -83,6 +84,7 @@
  * @author Will Schipp
  * @author Michael Minella
  * @author Mahmoud Ben Hassine
+ * @author Yanming Zhou
  *
  */
 @MessageEndpoint
@@ -286,7 +288,7 @@ private Set<StepExecution> pollReplies(final StepExecution managerStepExecution,
 
 	@SuppressWarnings("unchecked")
 	private Set<StepExecution> receiveReplies(PollableChannel currentReplyChannel) {
-		Message<Set<StepExecution>> message = (Message<Set<StepExecution>>) messagingGateway
+		Message<Collection<StepExecution>> message = (Message<Collection<StepExecution>>) messagingGateway
 			.receive(currentReplyChannel);
 
 		if (message == null) {
@@ -296,7 +298,8 @@ else if (logger.isDebugEnabled()) {
 			logger.debug("Received replies: " + message);
 		}
 
-		return new HashSet<>(message.getPayload());
+		Collection<StepExecution> payload = message.getPayload();
+		return payload instanceof Set ? (Set<StepExecution>) payload : new HashSet<>(message.getPayload());
 	}
 
 	private Message<StepExecutionRequest> createMessage(int sequenceNumber, int sequenceSize,