diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java
index c97915bd0..367dadd22 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java
@@ -60,7 +60,13 @@ public class JobFlowExecutor implements FlowExecutor {
 			StartLimitExceededException {
 		StepExecution stepExecution = stepHandler.handleStep(step, execution);
 		stepExecutionHolder.set(stepExecution);
-		return stepExecution == null ? ExitStatus.COMPLETED.getExitCode() : stepExecution.getExitStatus().getExitCode();
+		if (stepExecution == null) {
+			return  ExitStatus.COMPLETED.getExitCode();			
+		}
+		if (stepExecution.isTerminateOnly()) {
+			throw new JobInterruptedException("Step requested termination: "+stepExecution);
+		}
+		return stepExecution.getExitStatus().getExitCode();
 	}
 
 	public void abandonStepExecution() {
@@ -93,7 +99,7 @@ public class JobFlowExecutor implements FlowExecutor {
 		if (getStepExecution() != null && getStepExecution().getStatus() == BatchStatus.ABANDONED) {
 			/*
 			 * This is assumed to be the last step execution and it was marked
-			 * abandoned, so we are in a restart of a stopped step. 
+			 * abandoned, so we are in a restart of a stopped step.
 			 */
 			// TODO: mark the step execution in some more definitive way?
 			return true;
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobFailureTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobFailureTests.java
index 55a53ab1c..e9b686379 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobFailureTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobFailureTests.java
@@ -46,16 +46,18 @@ public class SimpleJobFailureTests {
 
 	@Test
 	public void testStepStatusUnknown() throws Exception {
-		job.setSteps(Arrays.<Step> asList(new StepSupport("step") {
+		job.setSteps(Arrays.<Step> asList(new StepSupport("step1") {
 			@Override
 			public void execute(StepExecution stepExecution) throws JobInterruptedException,
 					UnexpectedJobExecutionException {
 				// This is what happens if the repository meta-data cannot be updated
 				stepExecution.setStatus(BatchStatus.UNKNOWN);
+				stepExecution.setTerminateOnly();
 			}
-		}));
+		}, new StepSupport("step2")));
 		job.execute(execution);
 		assertEquals(BatchStatus.UNKNOWN, execution.getStatus());
+		assertEquals(1, execution.getStepExecutions().size());
 	}
 
 }
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java
index d48826d36..75148b3c9 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java
@@ -208,6 +208,31 @@ public class FlowJobTests {
 		assertEquals(JobInterruptedException.class, jobExecution.getFailureExceptions().get(0).getClass());
 	}
 
+	@Test
+	public void testUnknownStatusStopsJob() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1") {
+			@Override
+			public void execute(StepExecution stepExecution) throws JobInterruptedException {
+				stepExecution.setStatus(BatchStatus.UNKNOWN);
+				stepExecution.setTerminateOnly();
+				jobRepository.update(stepExecution);
+			}
+		}), "step2"));
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step2")), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow.setStateTransitions(transitions);
+		flow.afterPropertiesSet();
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+		job.execute(jobExecution);
+		assertEquals(BatchStatus.STOPPED, jobExecution.getStatus());
+		checkRepository(BatchStatus.STOPPED, ExitStatus.STOPPED);
+		assertEquals(1, jobExecution.getAllFailureExceptions().size());
+		assertEquals(JobInterruptedException.class, jobExecution.getFailureExceptions().get(0).getClass());
+	}
+
 	@Test
 	public void testInterruptedSplit() throws Exception {
 		SimpleFlow flow = new SimpleFlow("job");