IN PROGRESS - issue BATCH-677: Partition SPI. Added first draft of partitioning

spring-batch-core/src/test/java/org/springframework/batch/core/BatchStatusTests.java

@@ -39,6 +39,13 @@ public class BatchStatusTests {
 		assertEquals("FAILED", BatchStatus.FAILED.toString());
 	}
 
+	@Test
+	public void testMaxStatus() {
+		assertEquals(BatchStatus.FAILED, BatchStatus.max(BatchStatus.FAILED,BatchStatus.COMPLETED));
+		assertEquals(BatchStatus.FAILED, BatchStatus.max(BatchStatus.COMPLETED, BatchStatus.FAILED));
+		assertEquals(BatchStatus.FAILED, BatchStatus.max(BatchStatus.FAILED, BatchStatus.FAILED));
+	}
+
 	@Test
 	public void testGetStatus() {
 		assertEquals(BatchStatus.FAILED, BatchStatus.valueOf(BatchStatus.FAILED.toString()));

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

@@ -0,0 +1,57 @@
+package org.springframework.batch.core.partition;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemStream;
+import org.springframework.batch.item.ItemStreamException;
+
+/**
+ * {@link ItemReader} with hard-coded input data.
+ */
+public class ExampleItemReader implements ItemReader<String>, ItemStream {
+
+	private Log logger = LogFactory.getLog(getClass());
+
+	private String[] input = { "Hello", "world!", "Go", "on", "punk", "make", "my", "day!" };
+
+	private int index = 0;
+
+	public static volatile boolean fail = false;
+
+	/**
+	 * Reads next record from input
+	 */
+	public String read() throws Exception {
+		if (index >= input.length) {
+			return null;
+		}
+		logger.info(String.format("Processing input index=%s, item=%s, in (%s)", index, input[index], this));
+		if (fail && index == 4) {
+			synchronized (ExampleItemReader.class) {
+				if (fail) {
+					// Only fail once per flag setting...
+					fail = false;
+					logger.info(String.format("Throwing exception index=%s, item=%s, in (%s)", index, input[index],
+							this));
+					index++;
+					throw new RuntimeException("Planned failure");
+				}
+			}
+		}
+		return input[index++];
+	}
+
+	public void close(ExecutionContext executionContext) throws ItemStreamException {
+	}
+
+	public void open(ExecutionContext executionContext) throws ItemStreamException {
+		index = (int) executionContext.getLong("POSITION", 0);
+	}
+
+	public void update(ExecutionContext executionContext) throws ItemStreamException {
+		executionContext.putLong("POSITION", index);
+	}
+
+}

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

@@ -0,0 +1,70 @@
+package org.springframework.batch.core.partition;
+
+import static org.junit.Assert.*;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.batch.item.ExecutionContext;
+
+public class ExampleItemReaderTests {
+	
+	private ExampleItemReader reader = new ExampleItemReader();
+	
+	@Before
+	@After
+	public void ensureFailFlagUnset() {
+		ExampleItemReader.fail = false;
+	}
+
+	@Test
+	public void testRead() throws Exception {
+		int count = 0;
+		while (reader.read()!=null) {
+			count++;
+		}
+		assertEquals(8, count);
+	}
+
+	@Test
+	public void testOpen() throws Exception {
+		ExecutionContext context = new ExecutionContext();
+		for (int i=0; i<4; i++) {
+			reader.read();
+		}
+		reader.update(context);
+		reader.open(context);
+		int count = 0;
+		while (reader.read()!=null) {
+			count++;
+		}
+		assertEquals(4, count);		
+	}
+
+	@Test
+	public void testFailAndRestart() throws Exception {
+		ExecutionContext context = new ExecutionContext();
+		ExampleItemReader.fail = true;
+		for (int i=0; i<4; i++) {
+			reader.read();
+			reader.update(context);
+		}
+		try {
+			reader.read();
+			reader.update(context);
+			fail("Expected Exception");
+		}
+		catch (Exception e) {
+			// expected
+			assertEquals("Planned failure", e.getMessage());
+		}
+		assertFalse(ExampleItemReader.fail);
+		reader.open(context);
+		int count = 0;
+		while (reader.read()!=null) {
+			count++;
+		}
+		assertEquals(4, count);		
+	}
+
+}

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

@@ -0,0 +1,23 @@
+package org.springframework.batch.core.partition;
+
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.item.ItemWriter;
+
+/**
+ * Dummy {@link ItemWriter} which only logs data it receives.
+ */
+public class ExampleItemWriter implements ItemWriter<Object> {
+
+	private static final Log log = LogFactory.getLog(ExampleItemWriter.class);
+
+	/**
+	 * @see ItemWriter#write(List)
+	 */
+	public void write(List<? extends Object> data) throws Exception {
+		log.info(data);
+	}
+
+}

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

@@ -0,0 +1,94 @@
+/*
+ * Copyright 2006-2007 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import javax.sql.DataSource;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.JobParametersBuilder;
+import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+@ContextConfiguration(locations = "launch-context.xml")
+@RunWith(SpringJUnit4ClassRunner.class)
+public class RestartIntegrationTests {
+
+	@Autowired
+	private JobLauncher jobLauncher;
+
+	@Autowired
+	private Job job;
+
+	private SimpleJdbcTemplate jdbcTemplate;
+	
+	@Autowired
+	public void setDataSource(DataSource dataSource) {
+		this.jdbcTemplate = new SimpleJdbcTemplate(dataSource);
+	}
+
+	@Test
+	public void testSimpleProperties() throws Exception {
+		assertNotNull(jobLauncher);
+	}
+
+	@Before
+	@After
+	public void start() {
+		ExampleItemReader.fail = false;
+	}
+
+	@Test
+	public void testLaunchJob() throws Exception {
+
+		// Force failure in one of the parallel steps
+		ExampleItemReader.fail = true;
+		JobParameters jobParameters = new JobParametersBuilder().addString("restart", "yes").toJobParameters();
+
+		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%'");
+
+		JobExecution execution = jobLauncher.run(job, jobParameters);
+		assertEquals(BatchStatus.FAILED,execution.getStatus());
+		assertNotNull(jobLauncher.run(job, jobParameters));
+
+		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%'");
+
+		// Two attempts
+		assertEquals(2, afterMaster-beforeMaster);
+		// One failure and two successes
+		assertEquals(3, afterPartition-beforePartition);
+
+	}
+
+}

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

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2006-2007 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import javax.sql.DataSource;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+@ContextConfiguration(locations="launch-context.xml")
+@RunWith(SpringJUnit4ClassRunner.class)
+public class VanillaIntegrationTests {
+
+	@Autowired
+	private JobLauncher jobLauncher;
+
+	@Autowired
+	private Job job;
+
+	private SimpleJdbcTemplate jdbcTemplate;
+	
+	@Autowired
+	public void setDataSource(DataSource dataSource) {
+		this.jdbcTemplate = new SimpleJdbcTemplate(dataSource);
+	}
+
+	@Test
+	public void testSimpleProperties() throws Exception {
+		assertNotNull(jobLauncher);
+	}
+
+	@Test
+	public void testLaunchJob() throws Exception {
+		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%'");
+		assertNotNull(jobLauncher.run(job, new JobParameters()));
+		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%'");
+		assertEquals(1, afterMaster-beforeMaster);
+		// Should be same as grid size in step splitter
+		assertEquals(2, afterPartition-beforePartition);
+	}
+
+}

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

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2006-2007 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.support;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Collection;
+import java.util.Set;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.partition.PartitionHandler;
+import org.springframework.batch.core.partition.StepExecutionSplitter;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
+import org.springframework.batch.core.step.StepSupport;
+import org.springframework.batch.repeat.ExitStatus;
+import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
+
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class PartitionStepTests {
+	
+	private PartitionStep step = new PartitionStep();
+	private Step remote = new StepSupport("remote");
+	private JobRepository jobRepository;
+
+	@Before
+	public void setUp() throws Exception {
+		MapJobRepositoryFactoryBean factory = new MapJobRepositoryFactoryBean();
+		factory.setTransactionManager(new ResourcelessTransactionManager());
+		jobRepository = (JobRepository) factory.getObject();
+		step.setJobRepository(jobRepository);
+	}
+
+	@Test
+	public void testVanillaStepExecution() throws Exception {
+		step.setStepExecutionSplitter(new SimpleStepExecutionSplitter(jobRepository, remote));
+		step.setPartitionHandler(new PartitionHandler() {
+			public Collection<StepExecution> handle(StepExecutionSplitter stepSplitter, StepExecution stepExecution)
+					throws Exception {
+				Set<StepExecution> executions = stepSplitter.split(stepExecution, 2);
+				for (StepExecution execution : executions) {
+					execution.setStatus(BatchStatus.COMPLETED);
+					execution.setExitStatus(ExitStatus.FINISHED);
+				}
+				return executions;
+			}
+		});
+		step.afterPropertiesSet();
+		StepExecution stepExecution = new JobExecution(0L).createStepExecution("foo");
+		step.execute(stepExecution);
+		// one master and two workers
+		assertEquals(3, stepExecution.getJobExecution().getStepExecutions().size());
+		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
+	}
+
+	@Test
+	public void testFailedStepExecution() throws Exception {
+		step.setStepExecutionSplitter(new SimpleStepExecutionSplitter(jobRepository, remote));
+		step.setPartitionHandler(new PartitionHandler() {
+			public Collection<StepExecution> handle(StepExecutionSplitter stepSplitter, StepExecution stepExecution)
+					throws Exception {
+				Set<StepExecution> executions = stepSplitter.split(stepExecution, 2);
+				for (StepExecution execution : executions) {
+					execution.setStatus(BatchStatus.FAILED);
+					execution.setExitStatus(ExitStatus.FAILED);
+				}
+				return executions;
+			}
+		});
+		step.afterPropertiesSet();
+		StepExecution stepExecution = new JobExecution(0L).createStepExecution("foo");
+		step.execute(stepExecution);
+		// one master and two workers
+		assertEquals(3, stepExecution.getJobExecution().getStepExecutions().size());
+		assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
+	}
+
+}

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

@@ -0,0 +1,67 @@
+package org.springframework.batch.core.partition.support;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Collections;
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.partition.support.Partitioner;
+import org.springframework.batch.core.partition.support.SimpleStepExecutionSplitter;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
+import org.springframework.batch.core.step.tasklet.TaskletStep;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
+
+public class SimpleStepExecutionSplitterTests {
+
+	private Step step;
+
+	private JobRepository jobRepository;
+
+	private StepExecution stepExecution = new StepExecution("bar", new JobExecution(11L));
+
+	@Before
+	public void setUp() throws Exception {
+		step = new TaskletStep("step");
+		MapJobRepositoryFactoryBean factory = new MapJobRepositoryFactoryBean();
+		factory.setTransactionManager(new ResourcelessTransactionManager());
+		jobRepository = (JobRepository) factory.getObject();
+	}
+
+	@Test
+	public void testSimpleStepExecutionProviderJobRepositoryStep() throws Exception {
+		SimpleStepExecutionSplitter provider = new SimpleStepExecutionSplitter(jobRepository, step);
+		assertEquals(2, provider.split(stepExecution, 2).size());
+	}
+
+	@Test
+	public void testSimpleStepExecutionProviderJobRepositoryStepPartitioner() throws Exception {
+		final Map<String, ExecutionContext> map = Collections.singletonMap("foo", new ExecutionContext());
+		SimpleStepExecutionSplitter provider = new SimpleStepExecutionSplitter(jobRepository, step, new Partitioner() {
+			public Map<String, ExecutionContext> partition(int gridSize) {
+				return map;
+			}
+		});
+		assertEquals(1, provider.split(stepExecution, 2).size());
+	}
+
+	@Test
+	public void testRememberGridSize() throws Exception {
+		SimpleStepExecutionSplitter provider = new SimpleStepExecutionSplitter(jobRepository, step);
+		assertEquals(2, provider.split(stepExecution, 2).size());
+		assertEquals(2, provider.split(stepExecution, 3).size());
+	}
+
+	@Test
+	public void testGetStepName() {
+		SimpleStepExecutionSplitter provider = new SimpleStepExecutionSplitter(jobRepository, step);
+		assertEquals("step", provider.getStepName());
+	}
+
+}

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

@@ -0,0 +1,71 @@
+package org.springframework.batch.core.partition.support;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import org.junit.Test;
+import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.repeat.ExitStatus;
+
+public class StepExecutionAggregatorTests {
+
+	private StepExecutionAggregator aggregator = new StepExecutionAggregator();
+
+	private JobExecution jobExecution = new JobExecution(11L);
+
+	private StepExecution result = jobExecution.createStepExecution("aggregate");
+
+	private StepExecution stepExecution1 = jobExecution.createStepExecution("foo:1");
+
+	private StepExecution stepExecution2 = jobExecution.createStepExecution("foo:2");
+
+	@Test(expected = IllegalArgumentException.class)
+	public void testAggregateEmpty() {
+		aggregator.aggregate(result, Collections.<StepExecution> emptySet());
+	}
+
+	@Test(expected = IllegalArgumentException.class)
+	public void testAggregateNull() {
+		aggregator.aggregate(result, null);
+	}
+
+	@Test
+	public void testAggregateStatusSunnyDay() {
+		stepExecution1.setStatus(BatchStatus.COMPLETED);
+		stepExecution2.setStatus(BatchStatus.COMPLETED);
+		aggregator.aggregate(result, Arrays.<StepExecution> asList(stepExecution1, stepExecution2));
+		assertNotNull(result);
+		assertEquals(BatchStatus.COMPLETED, result.getStatus());
+	}
+
+	@Test
+	public void testAggregateStatusIncomplete() {
+		stepExecution1.setStatus(BatchStatus.COMPLETED);
+		stepExecution2.setStatus(BatchStatus.FAILED);
+		aggregator.aggregate(result, Arrays.<StepExecution> asList(stepExecution1, stepExecution2));
+		assertNotNull(result);
+		assertEquals(BatchStatus.FAILED, result.getStatus());
+	}
+
+	@Test
+	public void testAggregateExitStatusSunnyDay() {
+		stepExecution1.setExitStatus(ExitStatus.CONTINUABLE);
+		stepExecution2.setExitStatus(ExitStatus.FAILED);
+		aggregator.aggregate(result, Arrays.<StepExecution> asList(stepExecution1, stepExecution2));
+		assertNotNull(result);
+		assertEquals(ExitStatus.FAILED.and(ExitStatus.CONTINUABLE), result.getExitStatus());
+	}
+
+	@Test
+	public void testAggregateCommitCountSunnyDay() {
+		stepExecution1.setCommitCount(10);
+		stepExecution2.setCommitCount(5);
+		aggregator.aggregate(result, Arrays.<StepExecution> asList(stepExecution1, stepExecution2));
+		assertEquals(15, result.getCommitCount());
+	}
+}

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

@@ -0,0 +1,103 @@
+package org.springframework.batch.core.partition.support;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobExecutionException;
+import org.springframework.batch.core.JobInterruptedException;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.partition.StepExecutionSplitter;
+import org.springframework.batch.core.step.StepSupport;
+import org.springframework.batch.repeat.ExitStatus;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.core.task.TaskRejectedException;
+
+public class TaskExecutorPartitionHandlerTests {
+
+	private TaskExecutorPartitionHandler handler = new TaskExecutorPartitionHandler();
+
+	private int count = 0;
+
+	private StepExecution stepExecution = new StepExecution("step", new JobExecution(1L));
+
+	private StepExecutionSplitter stepExecutionSplitter = new StepExecutionSplitter() {
+
+		public String getStepName() {
+			return stepExecution.getStepName();
+		}
+
+		public Set<StepExecution> split(StepExecution stepExecution, int gridSize) throws JobExecutionException {
+			HashSet<StepExecution> result = new HashSet<StepExecution>();
+			for (int i = gridSize; i-- > 0;) {
+				result.add(stepExecution.getJobExecution().createStepExecution("foo" + i));
+			}
+			return result;
+		}
+	};
+
+	@Before
+	public void setUp() throws Exception {
+		handler.setStep(new StepSupport() {
+			@Override
+			public void execute(StepExecution stepExecution) throws JobInterruptedException {
+				count++;
+			}
+		});
+		handler.afterPropertiesSet();
+	}
+
+	@Test
+	public void testAfterPropertiesSet() throws Exception {
+		handler = new TaskExecutorPartitionHandler();
+		try {
+			handler.afterPropertiesSet();
+			fail("Expected IllegalArgumentException");
+		}
+		catch (IllegalArgumentException e) {
+			// expected
+			String message = e.getMessage();
+			assertTrue("Wrong message: " + message, message.contains("Step"));
+		}
+	}
+
+	@Test
+	public void testSetGridSize() throws Exception {
+		handler.setGridSize(2);
+		handler.handle(stepExecutionSplitter, stepExecution);
+		assertEquals(2, count);
+	}
+
+	@Test
+	public void testSetTaskExecutor() throws Exception {
+		handler.setTaskExecutor(new SimpleAsyncTaskExecutor());
+		handler.handle(stepExecutionSplitter, stepExecution);
+		assertEquals(1, count);
+	}
+
+	@Test
+	public void testTaskExecutorFailure() throws Exception {
+		handler.setGridSize(2);
+		handler.setTaskExecutor(new TaskExecutor() {
+			public void execute(Runnable task) {
+				if (count > 0) {
+					throw new TaskRejectedException("foo");
+				}
+				task.run();
+			}
+		});
+		Collection<StepExecution> executions = handler.handle(stepExecutionSplitter, stepExecution);
+		new StepExecutionAggregator().aggregate(stepExecution, executions);
+		assertEquals(1, count);
+		assertEquals(ExitStatus.FAILED.getExitCode(), stepExecution.getExitStatus().getExitCode());
+	}
+
+}

spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepContextRepeatCallbackTests.java

@@ -16,6 +16,8 @@
 package org.springframework.batch.core.scope;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 
 import org.junit.Test;
 import org.springframework.batch.core.JobExecution;
@@ -25,21 +27,46 @@ import org.springframework.batch.repeat.RepeatContext;
 
 /**
  * @author Dave Syer
- *
+ * 
  */
 public class StepContextRepeatCallbackTests {
+	
+	private StepExecution stepExecution = new StepExecution("foo", new JobExecution(0L), 123L);
+	private boolean addedAttribute = false;
+	private boolean removedAttribute = false;
 
 	@Test
 	public void testDoInIteration() throws Exception {
-		 StepContext stepContext = new StepContext(new StepExecution("foo", new JobExecution(0L), 123L));
-		StepContextRepeatCallback callback = new  StepContextRepeatCallback(stepContext ) {
+		StepContextRepeatCallback callback = new StepContextRepeatCallback(stepExecution) {
 			@Override
 			public ExitStatus doInStepContext(RepeatContext context, StepContext stepContext) throws Exception {
 				assertEquals(Long.valueOf(123), stepContext.getStepExecution().getId());
 				return ExitStatus.NOOP;
 			}
-		 };
-		 assertEquals(ExitStatus.NOOP, callback.doInIteration(null));
+		};
+		assertEquals(ExitStatus.NOOP, callback.doInIteration(null));
 	}
 
+	@Test
+	public void testUnfinishedWork() throws Exception {
+		StepContextRepeatCallback callback = new StepContextRepeatCallback(stepExecution) {
+			@Override
+			public ExitStatus doInStepContext(RepeatContext context, StepContext stepContext) throws Exception {
+				if (addedAttribute) {
+					removedAttribute = stepContext.hasAttribute("foo");
+					stepContext.removeAttribute("foo");
+				} else {
+					addedAttribute = true;
+					stepContext.setAttribute("foo", "bar");
+				}
+				return ExitStatus.NOOP;
+			}
+		};
+		callback.doInIteration(null);
+		assertTrue(addedAttribute);
+		callback.doInIteration(null);
+		assertTrue(removedAttribute);
+		callback.doInIteration(null);
+		assertFalse(removedAttribute);
+	}
 }

spring-batch-core/src/test/java/org/springframework/batch/core/step/AbstractStepTests.java

@@ -1,6 +1,7 @@
 package org.springframework.batch.core.step;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -132,6 +133,34 @@ public class AbstractStepTests {
 	public void setUp() throws Exception {
 		tested.setJobRepository(repository);
 	}
+	
+	@Test
+	public void testBeanName() throws Exception {
+		AbstractStep step = new AbstractStep() {
+			@Override
+			protected ExitStatus doExecute(StepExecution stepExecution) throws Exception {
+				return null;
+			}
+		};
+		assertNull(step.getName());
+		step.setBeanName("foo");
+		assertEquals("foo", step.getName());
+	}
+
+	@Test
+	public void testName() throws Exception {
+		AbstractStep step = new AbstractStep() {
+			@Override
+			protected ExitStatus doExecute(StepExecution stepExecution) throws Exception {
+				return null;
+			}
+		};
+		assertNull(step.getName());
+		step.setName("foo");
+		assertEquals("foo", step.getName());
+		step.setBeanName("bar");
+		assertEquals("foo", step.getName());
+	}
 
 	/**
 	 * Typical step execution scenario.

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTaskletTests.java

@@ -28,9 +28,9 @@ import org.junit.Before;
 import org.junit.Test;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.scope.ChunkContext;
 import org.springframework.batch.core.step.skip.ItemSkipPolicy;
 import org.springframework.batch.core.step.skip.SkipLimitExceededException;
-import org.springframework.batch.core.step.tasklet.BasicAttributeAccessor;
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemWriter;
@@ -112,7 +112,7 @@ public class FaultTolerantChunkOrientedTaskletTests {
 		handler = new FaultTolerantChunkOrientedTasklet<Integer, String>(itemReader, itemProcessor, itemWriter, chunkOperations,
 				retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy);
 		StepContribution contribution = new StepExecution("foo", null).createStepContribution();
-		handler.execute(contribution, new BasicAttributeAccessor());
+		handler.execute(contribution, new ChunkContext());
 		assertEquals(limit, contribution.getReadCount());
 	}
 
@@ -126,7 +126,7 @@ public class FaultTolerantChunkOrientedTaskletTests {
 				writeSkipPolicy, writeSkipPolicy);
 		chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1));
 		StepContribution contribution = new StepExecution("foo", null).createStepContribution();
-		BasicAttributeAccessor attributes = new BasicAttributeAccessor();
+		ChunkContext attributes = new ChunkContext();
 		try {
 			handler.execute(contribution, attributes);
 			fail("Expected SkipLimitExceededException");
@@ -148,7 +148,7 @@ public class FaultTolerantChunkOrientedTaskletTests {
 		}, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy);
 		chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1));
 		StepContribution contribution = new StepExecution("foo", null).createStepContribution();
-		BasicAttributeAccessor attributes = new BasicAttributeAccessor();
+		ChunkContext attributes = new ChunkContext();
 		try {
 			handler.execute(contribution, attributes);
 			fail("Expected RuntimeException");
@@ -174,7 +174,7 @@ public class FaultTolerantChunkOrientedTaskletTests {
 		}, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy);
 		chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(2));
 		StepContribution contribution = new StepExecution("foo", null).createStepContribution();
-		BasicAttributeAccessor attributes = new BasicAttributeAccessor();
+		ChunkContext attributes = new ChunkContext();
 
 		// Count to 3: (try + skip + skip)
 		for (int i = 0; i < 3; i++) {
@@ -193,7 +193,7 @@ public class FaultTolerantChunkOrientedTaskletTests {
 		// The last recovery for this chunk...
 		handler.execute(contribution, attributes);
 
-		attributes = new BasicAttributeAccessor();
+		attributes = new ChunkContext();
 		try {
 			handler.execute(contribution, attributes);
 			fail("Expected RuntimeException");
@@ -230,7 +230,7 @@ public class FaultTolerantChunkOrientedTaskletTests {
 				writeSkipPolicy);
 		chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(3));
 		StepContribution contribution = new StepExecution("foo", null).createStepContribution();
-		BasicAttributeAccessor attributes = new BasicAttributeAccessor();
+		ChunkContext attributes = new ChunkContext();
 
 		// try
 		try {
@@ -267,7 +267,7 @@ public class FaultTolerantChunkOrientedTaskletTests {
 				writeSkipPolicy);
 		chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(2));
 		StepContribution contribution = new StepExecution("foo", null).createStepContribution();
-		BasicAttributeAccessor attributes = new BasicAttributeAccessor();
+		ChunkContext attributes = new ChunkContext();
 
 		// Count to 3: (try + skip + try)
 		for (int i = 0; i < 2; i++) {
@@ -288,7 +288,7 @@ public class FaultTolerantChunkOrientedTaskletTests {
 		handler.execute(contribution, attributes);
 		assertEquals(2, chunk.getSkips().size());
 
-		attributes = new BasicAttributeAccessor();
+		attributes = new ChunkContext();
 		try {
 			handler.execute(contribution, attributes);
 			fail("Expected RuntimeException");