Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Updated reply channel in the MessageChannePartitionHandler

Browse Source 
When using the `MessageChannelPartitionHandler`, the reply channel is an
instance variable.  However, when the handle method is called, if a
reply channel is not injected, it will create a new one.  If multiple
jobs recycle the same instance of the partition handler, you may end up
having messages crossed/lost.  This PR removes the creation of the reply
channel for each call to the handle method and documents that this
comonent should be step scoped.

Resolves BATCH-2288
This commit is contained in:
Michael Minella
2016-10-13 12:42:39 -05:00
parent 0c466339e5
commit ad5348f32d
1 changed files with 11 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -54,6 +54,10 @@ import org.springframework.util.CollectionUtils;
* repository, we can poll the store to determine the state without the need of the slaves to formally respond.</li>
* </ul>
*
* Note: The reply channel for this is instance based. Sharing this component across
* multiple step instances may result in the crossing of messages. It's recommended that
* this component be step or job scoped.
*
* @author Dave Syer
* @author Will Schipp
* @author Michael Minella
@@ -102,6 +106,11 @@ public class MessageChannelPartitionHandler implements PartitionHandler, Initial
jobExplorerFactoryBean.afterPropertiesSet();
jobExplorer = jobExplorerFactoryBean.getObject();
}
if (!pollRepositoryForResults && replyChannel == null) {
replyChannel = new QueueChannel();
}//end if
}
/**
@@ -209,15 +218,9 @@ public class MessageChannelPartitionHandler implements PartitionHandler, Initial
int count = 0;
PollableChannel currentReplyChannel = replyChannel;
if (!pollRepositoryForResults && currentReplyChannel == null) {
currentReplyChannel = new QueueChannel();
}//end if
for (StepExecution stepExecution : split) {
Message<StepExecutionRequest> request = createMessage(count++, split.size(), new StepExecutionRequest(
stepName, stepExecution.getJobExecutionId(), stepExecution.getId()), currentReplyChannel);
stepName, stepExecution.getJobExecutionId(), stepExecution.getId()), replyChannel);
if (logger.isDebugEnabled()) {
logger.debug("Sending request: " + request);
}
@@ -225,7 +228,7 @@ public class MessageChannelPartitionHandler implements PartitionHandler, Initial
}
if(!pollRepositoryForResults) {
return receiveReplies(currentReplyChannel);
return receiveReplies(replyChannel);
}
else {
return pollReplies(masterStepExecution, split);