BATCH-1531: only re-compute partitions if not restarting, otherwise leave them empty because the normal step restart will take care of the state.

spring-batch-core/src/test/java/org/springframework/batch/core/partition/ExampleItemReader.java

@@ -18,13 +18,31 @@ public class ExampleItemReader implements ItemReader<String>, ItemStream {
 
 	private int index = 0;
 
+	private int min = 0;
+
+	private int max = Integer.MAX_VALUE;
+
 	public static volatile boolean fail = false;
 
+	/**
+	 * @param min the min to set
+	 */
+	public void setMin(int min) {
+		this.min = min;
+	}
+
+	/**
+	 * @param max the max to set
+	 */
+	public void setMax(int max) {
+		this.max = max;
+	}
+
 	/**
 	 * Reads next record from input
 	 */
 	public String read() throws Exception {
-		if (index >= input.length) {
+		if (index >= input.length || index >= max) {
 			return null;
 		}
 		logger.info(String.format("Processing input index=%s, item=%s, in (%s)", index, input[index], this));
@@ -47,7 +65,7 @@ public class ExampleItemReader implements ItemReader<String>, ItemStream {
 	}
 
 	public void open(ExecutionContext executionContext) throws ItemStreamException {
-		index = (int) executionContext.getLong("POSITION", 0);
+		index = (int) executionContext.getLong("POSITION", min);
 	}
 
 	public void update(ExecutionContext executionContext) throws ItemStreamException {

spring-batch-core/src/test/java/org/springframework/batch/core/partition/ExampleItemWriter.java

@@ -1,5 +1,6 @@
 package org.springframework.batch.core.partition;
 
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.commons.logging.Log;
@@ -9,15 +10,26 @@ import org.springframework.batch.item.ItemWriter;
 /**
  * Dummy {@link ItemWriter} which only logs data it receives.
  */
-public class ExampleItemWriter implements ItemWriter<Object> {
+public class ExampleItemWriter implements ItemWriter<String> {
 
 	private static final Log log = LogFactory.getLog(ExampleItemWriter.class);
+	
+	private static List<String> items = new ArrayList<String>();
+	
+	public static void clear() {
+		items.clear();
+	}
+
+	public static List<String> getItems() {
+		return items;
+	}
 
 	/**
 	 * @see ItemWriter#write(List)
 	 */
-	public void write(List<? extends Object> data) throws Exception {
+	public void write(List<? extends String> data) throws Exception {
 		log.info(data);
+		items.addAll(data);
 	}
 
 }

spring-batch-core/src/test/java/org/springframework/batch/core/partition/MinMaxPartitioner.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2006-2009 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.batch.core.partition;
+
+import java.util.Map;
+
+import org.springframework.batch.core.partition.support.SimplePartitioner;
+import org.springframework.batch.item.ExecutionContext;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class MinMaxPartitioner extends SimplePartitioner {
+
+	public Map<String, ExecutionContext> partition(int gridSize) {
+		Map<String, ExecutionContext> partition = super.partition(gridSize);
+		int total = 8; // The number of items in the ExampleItemReader
+		int range = total/gridSize;
+		int i = 0;
+		for (ExecutionContext context : partition.values()) {
+			 int min = (i++)*range;
+			 int max = Math.min(total, (min+1)*range);
+			 context.putInt("min", min);
+			 context.putInt("max", max);
+		}
+		return partition;
+	}
+
+}

spring-batch-core/src/test/java/org/springframework/batch/core/partition/RestartIntegrationTests.java

@@ -77,9 +77,16 @@ public class RestartIntegrationTests {
 		int beforeMaster = jdbcTemplate.queryForInt("SELECT COUNT(*) from BATCH_STEP_EXECUTION where STEP_NAME='step1:master'");
 		int beforePartition = jdbcTemplate.queryForInt("SELECT COUNT(*) from BATCH_STEP_EXECUTION where STEP_NAME like 'step1:partition%'");
 
+		ExampleItemWriter.clear();
 		JobExecution execution = jobLauncher.run(job, jobParameters);
 		assertEquals(BatchStatus.FAILED,execution.getStatus());
+		// Only 4 because the others were in the failed step execution
+		assertEquals(4, ExampleItemWriter.getItems().size());
+
+		ExampleItemWriter.clear();
 		assertNotNull(jobLauncher.run(job, jobParameters));
+		// Only 4 because the others were processed in the first attempt
+		assertEquals(4, ExampleItemWriter.getItems().size());
 
 		int afterMaster = jdbcTemplate.queryForInt("SELECT COUNT(*) from BATCH_STEP_EXECUTION where STEP_NAME='step1:master'");
 		int afterPartition = jdbcTemplate.queryForInt("SELECT COUNT(*) from BATCH_STEP_EXECUTION where STEP_NAME like 'step1:partition%'");