diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitter.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitter.java
index e25314213..1642d1974 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitter.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitter.java
@@ -94,9 +94,8 @@ public class SimpleStepExecutionSplitter implements StepExecutionSplitter, Initi
 
 	/**
 	 * Flag to indicate that the partition target step is allowed to start if an
-	 * execution is complete. Should be the same as the value that would be
-	 * returned by the {@link Step} itself from its own properties. Defaults to
-	 * false.
+	 * execution is complete. Defaults to the same value as the underlying step.
+	 * Set this manually to override the underlying step properties.
 	 * 
 	 * @see Step#isAllowStartIfComplete()
 	 * 
@@ -188,7 +187,8 @@ public class SimpleStepExecutionSplitter implements StepExecutionSplitter, Initi
 			// The context changed so we didn't already know the partitions
 			jobRepository.updateExecutionContext(stepExecution);
 			result = partitioner.partition(splitSize);
-		} else {
+		}
+		else {
 			result = new SimplePartitioner().partition(splitSize);
 		}
 
@@ -230,13 +230,28 @@ public class SimpleStepExecutionSplitter implements StepExecutionSplitter, Initi
 					+ "so it may be dangerous to proceed.  " + "Manual intervention is probably necessary.");
 		}
 
-		if (stepStatus == BatchStatus.COMPLETED && !allowStartIfComplete) {
-			// step is complete, false should be returned, indicating that the
-			// step should not be started
-			return false;
+		if (stepStatus == BatchStatus.COMPLETED) {
+			if (!allowStartIfComplete) {
+				// step is complete, false should be returned, indicating that
+				// the step should not be started
+				return false;
+			}
+			else {
+				return true;
+			}
 		}
 
-		return true;
+		if (stepStatus == BatchStatus.STOPPED || stepStatus == BatchStatus.FAILED) {
+			return true;
+		}
+
+		if (stepStatus == BatchStatus.STARTED || stepStatus == BatchStatus.STARTING || stepStatus == BatchStatus.STOPPING) {
+			throw new JobExecutionException("Cannot restart step from " + stepStatus + " status.  "
+					+ "The old execution may still be executing, so you may need to verify manually that this is the case.");
+		}
+
+		throw new JobExecutionException("Cannot restart step from " + stepStatus + " status.  "
+				+ "We believe the old execution was abandoned and therefore has been marked as un-restartable.");
 
 	}
 
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitterTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitterTests.java
index 2d695917a..265ab5ebc 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitterTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitterTests.java
@@ -2,13 +2,17 @@ package org.springframework.batch.core.partition.support;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 
 import java.util.Collections;
+import java.util.Date;
 import java.util.Map;
 import java.util.Set;
 
 import org.junit.Before;
 import org.junit.Test;
+import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.JobExecutionException;
 import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.StepExecution;
@@ -61,7 +65,9 @@ public class SimpleStepExecutionSplitterTests {
 	public void testRememberGridSize() throws Exception {
 		SimpleStepExecutionSplitter provider = new SimpleStepExecutionSplitter(jobRepository, step,
 				new SimplePartitioner());
-		assertEquals(2, provider.split(stepExecution, 2).size());
+		Set<StepExecution> split = provider.split(stepExecution, 2);
+		assertEquals(2, split.size());
+		stepExecution = update(split, stepExecution, BatchStatus.FAILED);
 		assertEquals(2, provider.split(stepExecution, 3).size());
 	}
 
@@ -72,4 +78,88 @@ public class SimpleStepExecutionSplitterTests {
 		assertEquals("step", provider.getStepName());
 	}
 
+	@Test
+	public void testUnkownStatus() throws Exception {
+		SimpleStepExecutionSplitter provider = new SimpleStepExecutionSplitter(jobRepository, step,
+				new SimplePartitioner());
+		Set<StepExecution> split = provider.split(stepExecution, 2);
+		assertEquals(2, split.size());
+		stepExecution = update(split, stepExecution, BatchStatus.UNKNOWN);
+		try {
+			provider.split(stepExecution, 2);
+		}
+		catch (JobExecutionException e) {
+			String message = e.getMessage();
+			assertTrue("Wrong message: " + message, message.contains("UNKNOWN"));
+		}
+	}
+
+	@Test
+	public void testCompleteStatus() throws Exception {
+		SimpleStepExecutionSplitter provider = new SimpleStepExecutionSplitter(jobRepository, step,
+				new SimplePartitioner());
+		Set<StepExecution> split = provider.split(stepExecution, 2);
+		assertEquals(2, split.size());
+		stepExecution = update(split, stepExecution, BatchStatus.COMPLETED);
+		// If already complete we don't execute again
+		assertEquals(0, provider.split(stepExecution, 2).size());
+	}
+
+	@Test
+	public void testIncompleteStatus() throws Exception {
+		SimpleStepExecutionSplitter provider = new SimpleStepExecutionSplitter(jobRepository, step,
+				new SimplePartitioner());
+		Set<StepExecution> split = provider.split(stepExecution, 2);
+		assertEquals(2, split.size());
+		stepExecution = update(split, stepExecution, BatchStatus.STARTED);
+		// If not already complete we don't execute again
+		try {
+			provider.split(stepExecution, 2);
+		}
+		catch (JobExecutionException e) {
+			String message = e.getMessage();
+			assertTrue("Wrong message: " + message, message.contains("STARTED"));
+		}
+	}
+
+	@Test
+	public void testAbandonedStatus() throws Exception {
+		SimpleStepExecutionSplitter provider = new SimpleStepExecutionSplitter(jobRepository, step,
+				new SimplePartitioner());
+		Set<StepExecution> split = provider.split(stepExecution, 2);
+		assertEquals(2, split.size());
+		stepExecution = update(split, stepExecution, BatchStatus.ABANDONED);
+		// If not already complete we don't execute again
+		try {
+			provider.split(stepExecution, 2);
+		}
+		catch (JobExecutionException e) {
+			String message = e.getMessage();
+			assertTrue("Wrong message: " + message, message.contains("ABANDONED"));
+		}
+	}
+
+	private StepExecution update(Set<StepExecution> split, StepExecution stepExecution, BatchStatus status)
+			throws Exception {
+
+		ExecutionContext executionContext = stepExecution.getExecutionContext();
+
+		for (StepExecution child : split) {
+			child.setEndTime(new Date());
+			child.setStatus(status);
+			jobRepository.update(child);
+		}
+
+		stepExecution.setEndTime(new Date());
+		stepExecution.setStatus(status);
+		jobRepository.update(stepExecution);
+
+		stepExecution = stepExecution.getJobExecution().createStepExecution(stepExecution.getStepName());
+		stepExecution.setExecutionContext(executionContext);
+
+		jobRepository.add(stepExecution);
+		return stepExecution;
+
+	}
+
 }