BATCH-2283: Updated MessageChannelPartitionHandler to return null when no partitions are specified

spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java

@@ -32,7 +32,7 @@ public class AsyncItemWriter<T> implements ItemWriter<Future<T>>, InitializingBe
 	}
 	
 	/**
-	 * @param delegate
+	 * @param delegate ItemWriter that does the actual writing of the Future results
 	 */
 	public void setDelegate(ItemWriter<T> delegate) {
 		this.delegate = delegate;

spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/JobLaunchRequest.java

@@ -31,8 +31,8 @@ public class JobLaunchRequest {
 	private final JobParameters jobParameters;
 
 	/**
-	 * @param job
-	 * @param jobParameters
+	 * @param job job to be launched
+	 * @param jobParameters parameters to run the job with
 	 */
 	public JobLaunchRequest(Job job, JobParameters jobParameters) {
 		super();

spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/JobLaunchingGateway.java

@@ -57,7 +57,7 @@ public class JobLaunchingGateway extends AbstractReplyProducingMessageHandler {
 	 * is a failure to start the job. The cause of the exception will be a
 	 * {@link JobExecutionException}.
 	 *
-	 * @throws MessageHandlingException
+	 * @throws MessageHandlingException when a job cannot be launched
 	 */
 	@Override
 	protected Object handleRequestMessage(Message<?> requestMessage) {

spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandler.java

@@ -35,7 +35,7 @@ public class JobLaunchingMessageHandler implements JobLaunchRequestHandler {
 	private final JobLauncher jobLauncher;
 
 	/**
-	 * @param jobLauncher
+	 * @param jobLauncher {@link org.springframework.batch.core.launch.JobLauncher} used to execute Spring Batch jobs
 	 */
 	public JobLaunchingMessageHandler(JobLauncher jobLauncher) {
 		super();

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

@@ -18,6 +18,7 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
 
 import java.util.Collection;
 import java.util.List;
@@ -119,6 +120,11 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
 			StepExecution masterStepExecution) throws Exception {
 
 		Set<StepExecution> split = stepExecutionSplitter.split(masterStepExecution, gridSize);
+
+		if(CollectionUtils.isEmpty(split)) {
+			return null;
+		}
+
 		int count = 0;
 
 		if (replyChannel == null) {

spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java

@@ -117,7 +117,7 @@ public class ChunkMessageItemWriterIntegrationTests {
 		factory.setItemReader(new ListItemReader<String>(Arrays.asList(StringUtils
 				.commaDelimitedListToStringArray("1,2,3,4,5,6"))));
 
-		Step step = (Step) factory.getObject();
+		Step step = factory.getObject();
 
 		StepExecution stepExecution = getStepExecution(step);
 		step.execute(stepExecution);
@@ -135,7 +135,7 @@ public class ChunkMessageItemWriterIntegrationTests {
 		factory.setItemReader(new ListItemReader<String>(Arrays.asList(StringUtils
 				.commaDelimitedListToStringArray("1,2,3,4,5,6"))));
 
-		Step step = (Step) factory.getObject();
+		Step step = factory.getObject();
 
 		StepExecution stepExecution = getStepExecution(step);
 
@@ -160,7 +160,7 @@ public class ChunkMessageItemWriterIntegrationTests {
 		factory.setItemReader(new ListItemReader<String>(Arrays.asList(StringUtils
 				.commaDelimitedListToStringArray("1,2,3,4,5,6"))));
 
-		Step step = (Step) factory.getObject();
+		Step step = factory.getObject();
 
 		StepExecution stepExecution = getStepExecution(step);
 
@@ -207,7 +207,7 @@ public class ChunkMessageItemWriterIntegrationTests {
 				.commaDelimitedListToStringArray("1,fail,3,4,5,6"))));
 		factory.setCommitInterval(2);
 
-		Step step = (Step) factory.getObject();
+		Step step = factory.getObject();
 
 		StepExecution stepExecution = getStepExecution(step);
 		step.execute(stepExecution);
@@ -233,7 +233,7 @@ public class ChunkMessageItemWriterIntegrationTests {
 		factory.setItemReader(new ListItemReader<String>(Arrays.asList(StringUtils
 				.commaDelimitedListToStringArray("1,2,3,4,5,6"))));
 
-		Step step = (Step) factory.getObject();
+		Step step = factory.getObject();
 
 		StepExecution stepExecution = getStepExecution(step);
 
@@ -271,7 +271,7 @@ public class ChunkMessageItemWriterIntegrationTests {
 		factory.setItemReader(new ListItemReader<String>(Arrays.asList(StringUtils
 				.commaDelimitedListToStringArray("wait,fail,3,4,5,6"))));
 
-		Step step = (Step) factory.getObject();
+		Step step = factory.getObject();
 
 		StepExecution stepExecution = getStepExecution(step);
 		step.execute(stepExecution);

spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandlerTests.java

@@ -1,6 +1,7 @@
 package org.springframework.batch.integration.partition;
 
 import org.junit.Test;
+import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.partition.StepExecutionSplitter;
 import org.springframework.integration.MessageTimeoutException;
@@ -10,10 +11,13 @@ import org.springframework.messaging.PollableChannel;
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.anyObject;
+import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -23,10 +27,25 @@ import static org.mockito.Mockito.when;
  * @author Michael Minella
  *
  */
-public class MessageChannelPartitionHandlerTest {
+@SuppressWarnings("raw")
+public class MessageChannelPartitionHandlerTests {
 
 	private MessageChannelPartitionHandler messageChannelPartitionHandler;
 
+	@Test
+	public void testNoPartitions() throws Exception {
+		//execute with no default set
+		messageChannelPartitionHandler = new MessageChannelPartitionHandler();
+		//mock
+		StepExecution masterStepExecution = mock(StepExecution.class);
+		StepExecutionSplitter stepExecutionSplitter = mock(StepExecutionSplitter.class);
+
+		//execute
+		Collection<StepExecution> executions = messageChannelPartitionHandler.handle(stepExecutionSplitter, masterStepExecution);
+		//verify
+		assertNull(executions);
+	}
+
 	@Test
 	public void testHandleNoReply() throws Exception {
 		//execute with no default set
@@ -37,6 +56,9 @@ public class MessageChannelPartitionHandlerTest {
 		MessagingTemplate operations = mock(MessagingTemplate.class);
 		Message message = mock(Message.class);
 		//when
+		HashSet<StepExecution> stepExecutions = new HashSet<StepExecution>();
+		stepExecutions.add(new StepExecution("step1", new JobExecution(5l)));
+		when(stepExecutionSplitter.split((StepExecution) anyObject(), eq(1))).thenReturn(stepExecutions);
 		when(message.getPayload()).thenReturn(Collections.emptyList());
 		when(operations.receive((PollableChannel) anyObject())).thenReturn(message);
 		//set
@@ -60,6 +82,9 @@ public class MessageChannelPartitionHandlerTest {
 		Message message = mock(Message.class);
 		PollableChannel replyChannel = mock(PollableChannel.class);
 		//when
+		HashSet<StepExecution> stepExecutions = new HashSet<StepExecution>();
+		stepExecutions.add(new StepExecution("step1", new JobExecution(5l)));
+		when(stepExecutionSplitter.split((StepExecution) anyObject(), eq(1))).thenReturn(stepExecutions);
 		when(message.getPayload()).thenReturn(Collections.emptyList());
 		when(operations.receive(replyChannel)).thenReturn(message);
 		//set
@@ -84,6 +109,9 @@ public class MessageChannelPartitionHandlerTest {
 		MessagingTemplate operations = mock(MessagingTemplate.class);
 		Message message = mock(Message.class);
 		//when
+		HashSet<StepExecution> stepExecutions = new HashSet<StepExecution>();
+		stepExecutions.add(new StepExecution("step1", new JobExecution(5l)));
+		when(stepExecutionSplitter.split((StepExecution) anyObject(), eq(1))).thenReturn(stepExecutions);
 		when(message.getPayload()).thenReturn(Collections.emptyList());
 		//set
 		messageChannelPartitionHandler.setMessagingOperations(operations);