diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java b/spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java
index f92712765..a669b606b 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java
@@ -16,7 +16,6 @@
 
 package org.springframework.batch.core;
 
-
 /**
  * Enumeration representing the status of a an Execution.
  * 
@@ -26,16 +25,22 @@ package org.springframework.batch.core;
 
 public enum BatchStatus {
 
-	COMPLETED, STARTED, STARTING, PAUSED, FAILED, STOPPING, STOPPED, UNKNOWN;
-	
+	/**
+	 * The order of the status values is significant. An execution is expected
+	 * to move up from lowest to highest, hopefully stopping at COMPLETED. After
+	 * COMPLETED, higher values signify more serious failure.
+	 */
+	STARTING, STARTED, COMPLETED, PAUSED, FAILED, STOPPING, STOPPED, UNKNOWN;
+
 	public static BatchStatus max(BatchStatus status1, BatchStatus status2) {
-		if (status1.compareTo(status2)<0) {
+		if (status1.compareTo(status2) < 0) {
 			return status2;
 		}
-		if (status1.compareTo(status2)>0) {
+		if (status1.compareTo(status2) > 0) {
 			return status1;
 		}
-		else return status1;
+		else
+			return status1;
 	}
 
 }
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java
index 5550facc6..c88a153d2 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java
@@ -48,13 +48,13 @@ public class JobExecution extends Entity {
 	private volatile Date createTime = new Date(System.currentTimeMillis());
 
 	private volatile Date endTime = null;
-	
+
 	private volatile Date lastUpdated = null;
 
 	private volatile ExitStatus exitStatus = ExitStatus.UNKNOWN;
 
 	private volatile ExecutionContext executionContext = new ExecutionContext();
-	
+
 	private transient volatile List<Throwable> failureExceptions = new ArrayList<Throwable>();
 
 	/**
@@ -76,8 +76,8 @@ public class JobExecution extends Entity {
 	public JobExecution(JobInstance job) {
 		this(job, null);
 	}
-	
-	public JobExecution(Long id){
+
+	public JobExecution(Long id) {
 		super(id);
 	}
 
@@ -88,7 +88,7 @@ public class JobExecution extends Entity {
 	public void setJobInstance(JobInstance jobInstance) {
 		this.jobInstance = jobInstance;
 	}
-	
+
 	public void setEndTime(Date endTime) {
 		this.endTime = endTime;
 	}
@@ -105,10 +105,28 @@ public class JobExecution extends Entity {
 		return status;
 	}
 
+	/**
+	 * Set the value of the status field.
+	 * 
+	 * @param status the status to set
+	 */
 	public void setStatus(BatchStatus status) {
 		this.status = status;
 	}
 
+	/**
+	 * Upgrade the status field if the provided value is greater than the
+	 * existing one. Clients using this method to set the status can be sure
+	 * that they don't overwrite a failed status with an successful one.
+	 * 
+	 * @param status the new status value
+	 */
+	public void upgradeStatus(BatchStatus status) {
+		if (status.compareTo(this.status) > 0) {
+			this.status = status;
+		}
+	}
+
 	/**
 	 * Convenience getter for for the id of the enclosing job. Useful for DAO
 	 * implementations.
@@ -194,12 +212,14 @@ public class JobExecution extends Entity {
 	}
 
 	/**
-	 * Signal that this job execution wishes to be paused.
+	 * Signal that this job execution wishes to be paused. Uses
+	 * {@link #upgradeStatus(BatchStatus)} so that a failed execution stays
+	 * failed.
 	 */
 	public void pause() {
-		status = BatchStatus.PAUSED;
+		upgradeStatus(BatchStatus.PAUSED);
 	}
-	
+
 	/**
 	 * Test if the {@link JobExecution} has been paused.
 	 * 
@@ -208,7 +228,7 @@ public class JobExecution extends Entity {
 	 * @return true if this instance is paused
 	 */
 	public boolean isPaused() {
-		return status==BatchStatus.PAUSED;
+		return status == BatchStatus.PAUSED;
 	}
 
 	/**
@@ -253,8 +273,8 @@ public class JobExecution extends Entity {
 	}
 
 	/**
-	 * Get the date representing the last time this JobExecution was updated in the
-	 * JobRepository.
+	 * Get the date representing the last time this JobExecution was updated in
+	 * the JobRepository.
 	 * 
 	 * @return Date representing the last time this JobExecution was updated.
 	 */
@@ -270,39 +290,40 @@ public class JobExecution extends Entity {
 	public void setLastUpdated(Date lastUpdated) {
 		this.lastUpdated = lastUpdated;
 	}
-	
+
 	public List<Throwable> getFailureExceptions() {
 		return failureExceptions;
 	}
-	
+
 	/**
-	 * Add the provided throwable to the failure exception list. 
+	 * Add the provided throwable to the failure exception list.
 	 * 
 	 * @param t
 	 */
-	public void addFailureException(Throwable t){
+	public void addFailureException(Throwable t) {
 		this.failureExceptions.add(t);
 	}
-	
+
 	/**
 	 * Return all failure causing exceptions for this JobExecution, including
 	 * step executions.
 	 * 
-	 * @return List<Throwable> containing all exceptions causing failure for this
-	 * JobExecution.
+	 * @return List<Throwable> containing all exceptions causing failure for
+	 * this JobExecution.
 	 */
-	public List<Throwable> getAllFailureExceptions(){
-		
+	public List<Throwable> getAllFailureExceptions() {
+
 		Set<Throwable> allExceptions = new HashSet<Throwable>(failureExceptions);
-		for(StepExecution stepExecution: stepExecutions){
+		for (StepExecution stepExecution : stepExecutions) {
 			allExceptions.addAll(stepExecution.getFailureExceptions());
 		}
-		
+
 		return new ArrayList<Throwable>(allExceptions);
 	}
 
 	/**
-	 * Deserialise and ensure transient fields are re-instantiated when read back
+	 * Deserialise and ensure transient fields are re-instantiated when read
+	 * back
 	 */
 	private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException {
 		stream.defaultReadObject();
@@ -311,10 +332,13 @@ public class JobExecution extends Entity {
 
 	/*
 	 * (non-Javadoc)
+	 * 
 	 * @see org.springframework.batch.core.domain.Entity#toString()
 	 */
 	public String toString() {
-		return super.toString() + String.format(", startTime=%s, endTime=%s, lastUpdated=%s, status=%s, exitStatus=%s, job=[%s]",startTime,endTime,lastUpdated,status,exitStatus,jobInstance);
+		return super.toString()
+				+ String.format(", startTime=%s, endTime=%s, lastUpdated=%s, status=%s, exitStatus=%s, job=[%s]",
+						startTime, endTime, lastUpdated, status, exitStatus, jobInstance);
 	}
 
 }
\ No newline at end of file
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/DecisionState.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/DecisionState.java
index 2942839fa..00b8647cb 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/DecisionState.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/DecisionState.java
@@ -20,7 +20,7 @@ public class DecisionState extends AbstractState<JobFlowExecutor> {
 
 	@Override
 	public String handle(JobFlowExecutor context) throws Exception {
-		return decider.decide(context.getJobExecution());
+		return decider.decide(context.getJobExecution(), context.getStepExecution());
 	}
 
 }
\ No newline at end of file
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/EndState.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/EndState.java
index b0038003e..6f731bfa6 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/EndState.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/EndState.java
@@ -38,7 +38,7 @@ public class EndState extends AbstractState<JobFlowExecutor> {
 		// If there are no step executions, then we are at the beginning of a
 		// restart
 		if (!jobExecution.getStepExecutions().isEmpty()) {
-			jobExecution.setStatus(status);
+			jobExecution.upgradeStatus(status);
 		}
 		return FlowExecution.COMPLETED;
 	}
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java
index 1556acb37..246091443 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java
@@ -26,6 +26,7 @@ import org.springframework.batch.core.repository.JobRestartException;
 import org.springframework.batch.flow.Flow;
 import org.springframework.batch.flow.FlowExecution;
 import org.springframework.batch.flow.FlowExecutionException;
+import org.springframework.batch.flow.FlowExecutionListenerSupport;
 import org.springframework.util.Assert;
 
 
@@ -65,15 +66,7 @@ public class FlowJob extends AbstractJob {
 	@Override
 	protected StepExecution doExecute(final JobExecution execution) throws JobExecutionException {
 		try {
-			FlowExecution result = flow.start(new JobFlowExecutor() {
-				public String executeStep(Step step) throws JobInterruptedException, JobRestartException, StartLimitExceededException {
-					StepExecution stepExecution = handleStep(step, execution);
-					return stepExecution==null ? FlowExecution.COMPLETED : stepExecution.getExitStatus().getExitCode();
-				}
-				public JobExecution getJobExecution() {
-					return execution;
-				}
-			});
+			FlowExecution result = flow.start(new SimpleJobFlowExecutor(execution));
 			return getLastStepExecution(execution, result);
 		}
 		catch (FlowExecutionException e) {
@@ -128,4 +121,42 @@ public class FlowJob extends AbstractJob {
 		return first.getEndTime().after(second.getEndTime());
 	}
 
+	/**
+	 * @author Dave Syer
+	 *
+	 */
+	private class SimpleJobFlowExecutor extends FlowExecutionListenerSupport implements JobFlowExecutor {
+
+		private final ThreadLocal<StepExecution> stepExecutionHolder = new ThreadLocal<StepExecution>();
+		private final JobExecution execution;
+
+		/**
+		 * @param execution
+		 */
+		private SimpleJobFlowExecutor(JobExecution execution) {
+			this.execution = execution;
+			stepExecutionHolder.set(null);
+		}
+
+		public String executeStep(Step step) throws JobInterruptedException, JobRestartException, StartLimitExceededException {
+			StepExecution stepExecution = handleStep(step, execution);
+			stepExecutionHolder.set(stepExecution);
+			return stepExecution==null ? FlowExecution.COMPLETED : stepExecution.getExitStatus().getExitCode();
+		}
+
+		public JobExecution getJobExecution() {
+			return execution;
+		}
+		
+		public StepExecution getStepExecution() {
+			return stepExecutionHolder.get();
+		}
+		
+		@Override
+		public void close(FlowExecution result) {
+			stepExecutionHolder.set(null);
+		}
+
+	}
+
 }
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobExecutionDecider.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobExecutionDecider.java
index a154c459f..6f476d493 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobExecutionDecider.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobExecutionDecider.java
@@ -16,6 +16,7 @@
 package org.springframework.batch.core.job.flow;
 
 import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.StepExecution;
 
 /**
  * @author Dave Syer
@@ -29,8 +30,9 @@ public interface JobExecutionDecider {
 	 * determine the next step in the job.
 	 * 
 	 * @param jobExecution a job execution
+	 * @param stepExecution the latest step execution (may be null)
 	 * @return the exit status code
 	 */
-	String decide(JobExecution jobExecution);
+	String decide(JobExecution jobExecution, StepExecution stepExecution);
 
 }
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 5a9702e66..86314cb92 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
@@ -19,6 +19,7 @@ import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.JobInterruptedException;
 import org.springframework.batch.core.StartLimitExceededException;
 import org.springframework.batch.core.Step;
+import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.repository.JobRestartException;
 import org.springframework.batch.flow.Flow;
 
@@ -45,4 +46,9 @@ public interface JobFlowExecutor {
 	 */
 	JobExecution getJobExecution();
 
+	/**
+	 * @return the latest {@link StepExecution} or null if there is none
+	 */
+	StepExecution getStepExecution();
+
 }
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/JobExecutionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/JobExecutionTests.java
index ee1cc1beb..7a991db69 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/JobExecutionTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/JobExecutionTests.java
@@ -92,6 +92,38 @@ public class JobExecutionTests {
 		assertEquals(BatchStatus.COMPLETED, execution.getStatus());
 	}
 
+	/**
+	 * Test method for
+	 * {@link org.springframework.batch.core.JobExecution#getStatus()}.
+	 */
+	@Test
+	public void testUpgradeStatus() {
+		assertEquals(BatchStatus.STARTING, execution.getStatus());
+		execution.upgradeStatus(BatchStatus.COMPLETED);
+		assertEquals(BatchStatus.COMPLETED, execution.getStatus());
+	}
+
+	/**
+	 * Test method for
+	 * {@link org.springframework.batch.core.JobExecution#pause()}.
+	 */
+	@Test
+	public void testPause() {
+		execution.pause();
+		assertEquals(BatchStatus.PAUSED, execution.getStatus());
+	}
+
+	/**
+	 * Test method for
+	 * {@link org.springframework.batch.core.JobExecution#getStatus()}.
+	 */
+	@Test
+	public void testDowngradeStatus() {
+		execution.setStatus(BatchStatus.FAILED);
+		execution.upgradeStatus(BatchStatus.COMPLETED);
+		assertEquals(BatchStatus.FAILED, execution.getStatus());
+	}
+
 	/**
 	 * Test method for
 	 * {@link org.springframework.batch.core.JobExecution#getJobId()}.
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/DecisionJobParserTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/DecisionJobParserTests.java
index 615dbe5d9..08d2e9d61 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/DecisionJobParserTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/DecisionJobParserTests.java
@@ -25,6 +25,7 @@ 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.StepExecution;
 import org.springframework.batch.core.job.flow.JobExecutionDecider;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
@@ -64,7 +65,7 @@ public class DecisionJobParserTests {
 	}
 	
 	public static class TestDecider implements JobExecutionDecider {
-		public String decide(JobExecution jobExecution) {
+		public String decide(JobExecution jobExecution, StepExecution stepExecution) {
 			return "FOO";
 		}
 	}
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StopJobParserTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StopJobParserTests.java
index c4e858e89..92548f461 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StopJobParserTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StopJobParserTests.java
@@ -25,6 +25,7 @@ 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.StepExecution;
 import org.springframework.batch.core.job.flow.JobExecutionDecider;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
@@ -67,7 +68,7 @@ public class StopJobParserTests {
 	}
 
 	public static class TestDecider implements JobExecutionDecider {
-		public String decide(JobExecution jobExecution) {
+		public String decide(JobExecution jobExecution, StepExecution stepExecution) {
 			return "FOO";
 		}
 	}
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/EndStateTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/EndStateTests.java
new file mode 100644
index 000000000..b33ebd9b3
--- /dev/null
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/EndStateTests.java
@@ -0,0 +1,103 @@
+/*
+ * 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.job.flow;
+
+import static org.junit.Assert.*;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.JobExecution;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class EndStateTests {
+
+	private JobExecution jobExecution;
+	
+	@Before
+	public void setUp() {
+		jobExecution = new JobExecution(0L);
+	}
+
+	/**
+	 * Test method for {@link EndState#handle(JobFlowExecutor)}.
+	 * @throws Exception 
+	 */
+	@Test
+	public void testHandleRestartSunnyDay() throws Exception {
+
+		BatchStatus status = jobExecution.getStatus();
+	
+		EndState state = new EndState(BatchStatus.UNKNOWN, "end");
+		state.handle(new JobFlowExecutorSupport() {
+			@Override
+			public JobExecution getJobExecution() {
+				return jobExecution;
+			}
+		});
+		
+		assertEquals(status, jobExecution.getStatus());
+
+	}
+
+	/**
+	 * Test method for {@link EndState#handle(JobFlowExecutor)}.
+	 * @throws Exception 
+	 */
+	@Test
+	public void testHandleOngoingSunnyDay() throws Exception {
+
+		jobExecution.createStepExecution("foo");
+	
+		EndState state = new EndState(BatchStatus.UNKNOWN, "end");
+		state.handle(new JobFlowExecutorSupport() {
+			@Override
+			public JobExecution getJobExecution() {
+				return jobExecution;
+			}
+		});
+		
+		assertEquals(BatchStatus.UNKNOWN, jobExecution.getStatus());
+
+	}
+
+	/**
+	 * Test method for {@link EndState#handle(JobFlowExecutor)}.
+	 * @throws Exception 
+	 */
+	@Test
+	public void testHandleOngoingAttemptedDowngrade() throws Exception {
+
+		jobExecution.setStatus(BatchStatus.FAILED);
+		jobExecution.createStepExecution("foo");
+	
+		EndState state = new EndState(BatchStatus.COMPLETED, "end");
+		state.handle(new JobFlowExecutorSupport() {
+			@Override
+			public JobExecution getJobExecution() {
+				return jobExecution;
+			}
+		});
+		
+		// Can't downgrade a status - if it failed then it failed
+		assertEquals(BatchStatus.FAILED, jobExecution.getStatus());
+
+	}
+
+}
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 8debfa572..d7a7ba3da 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
@@ -16,6 +16,7 @@
 package org.springframework.batch.core.job.flow;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.fail;
 
 import java.util.ArrayList;
@@ -214,12 +215,15 @@ public class FlowJobTests {
 
 	@Test
 	public void testDecisionFlow() throws Throwable {
+
 		SimpleFlow<JobFlowExecutor> flow = new SimpleFlow<JobFlowExecutor>("job");
 		JobExecutionDecider decider = new JobExecutionDecider() {
-			public String decide(JobExecution jobExecution) {
+			public String decide(JobExecution jobExecution, StepExecution stepExecution) {
+				assertNotNull(stepExecution);
 				return "SWITCH";
 			}
 		};
+
 		Collection<StateTransition<JobFlowExecutor>> transitions = new ArrayList<StateTransition<JobFlowExecutor>>();
 		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "*", "decision"));
 		transitions.add(StateTransition.createStateTransition(new DecisionState(decider, "decision"), "*", "step2"));
@@ -228,14 +232,17 @@ public class FlowJobTests {
 		transitions.add(StateTransition.createEndStateTransition(new StepState(new StubStep("step2")), "*"));
 		transitions.add(StateTransition.createEndStateTransition(new StepState(new StubStep("step3")), "*"));
 		flow.setStateTransitions(transitions);
+
 		job.setFlow(flow);
 		StepExecution stepExecution = job.doExecute(jobExecution);
 		if (!jobExecution.getAllFailureExceptions().isEmpty()) {
 			throw jobExecution.getAllFailureExceptions().get(0);
 		}
+
 		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
 		assertEquals(2, jobExecution.getStepExecutions().size());
 		assertEquals("step3", stepExecution.getStepName());
+
 	}
 
 	@Test
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/JobFlowExecutorSupport.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/JobFlowExecutorSupport.java
new file mode 100644
index 000000000..279a3a7ba
--- /dev/null
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/JobFlowExecutorSupport.java
@@ -0,0 +1,45 @@
+/*
+ * 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.job.flow;
+
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobInterruptedException;
+import org.springframework.batch.core.StartLimitExceededException;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.repository.JobRestartException;
+import org.springframework.batch.flow.FlowExecution;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class JobFlowExecutorSupport implements JobFlowExecutor {
+
+	public String executeStep(Step step) throws JobInterruptedException, JobRestartException,
+			StartLimitExceededException {
+		return FlowExecution.COMPLETED;
+	}
+
+	public JobExecution getJobExecution() {
+		return null;
+	}
+
+	public StepExecution getStepExecution() {
+		return null;
+	}
+
+}
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecutionListener.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecutionListener.java
new file mode 100644
index 000000000..26da76238
--- /dev/null
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecutionListener.java
@@ -0,0 +1,29 @@
+/*
+ * 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.flow;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public interface FlowExecutionListener {
+
+	/**
+	 * @param result
+	 */
+	void close(FlowExecution result);
+
+}
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecutionListenerSupport.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecutionListenerSupport.java
new file mode 100644
index 000000000..289758fee
--- /dev/null
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecutionListenerSupport.java
@@ -0,0 +1,31 @@
+/*
+ * 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.flow;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class FlowExecutionListenerSupport implements FlowExecutionListener {
+
+	/**
+	 * No-op implementation.
+	 * @see FlowExecutionListener#close(FlowExecution)
+	 */
+	public void close(FlowExecution result) {
+	}
+
+}
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/SimpleFlow.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/SimpleFlow.java
index d90be80bd..4a6e326ab 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/SimpleFlow.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/SimpleFlow.java
@@ -111,21 +111,37 @@ public class SimpleFlow<T> implements Flow<T>, InitializingBean {
 	 * @see Flow#resume(String, Object)
 	 */
 	public FlowExecution resume(String stateName, T context) throws FlowExecutionException {
+
 		String status = FlowExecution.UNKNOWN;
 		State<T> state = stateMap.get(stateName);
+
+		FlowExecutionListener listener = new FlowExecutionListenerSupport();
+		if (context instanceof FlowExecutionListener) {
+			listener = (FlowExecutionListener)context;
+		}
+
 		// Terminate if there are no more states
 		while (state != null) {
+
 			stateName = state.getName();
+
 			try {
 				status = state.handle(context);
 			}
 			catch (Exception e) {
+				listener.close(new FlowExecution(stateName, status));
 				throw new FlowExecutionException(String.format("Ended flow=%s at state=%s with exception", name,
 						stateName), e);
 			}
+
 			state = nextState(stateName, status);
+
 		}
-		return new FlowExecution(stateName, status);
+
+		FlowExecution result = new FlowExecution(stateName, status);
+		listener.close(result);
+		return result; 
+
 	}
 
 	/**
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/BasicFlowTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/BasicFlowTests.java
index 34105d5b5..5e5783169 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/BasicFlowTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/BasicFlowTests.java
@@ -22,6 +22,7 @@ import static org.junit.Assert.fail;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 
 import org.junit.Test;
 import org.springframework.batch.flow.FlowExecution;
@@ -35,34 +36,34 @@ import org.springframework.batch.flow.StateTransition;
  */
 public class BasicFlowTests {
 
-	private SimpleFlow<String> flow = new SimpleFlow<String>("job");
+	private SimpleFlow<Object> flow = new SimpleFlow<Object>("job");
 
-	private String executor = "data";
+	private Object executor = "data";
 
 	@Test(expected = IllegalArgumentException.class)
 	public void testEmptySteps() throws Exception {
-		flow.setStateTransitions(Collections.<StateTransition<String>> emptySet());
+		flow.setStateTransitions(Collections.<StateTransition<Object>> emptySet());
 		flow.afterPropertiesSet();
 	}
 
 	@Test(expected = IllegalArgumentException.class)
 	public void testNoNextStepSpecified() throws Exception {
-		flow.setStateTransitions(Collections.singleton(StateTransition.createStateTransition(new StateSupport<String>(
+		flow.setStateTransitions(Collections.singleton(StateTransition.createStateTransition(new StateSupport<Object>(
 				"step"), "foo")));
 		flow.afterPropertiesSet();
 	}
 
 	@Test(expected = IllegalArgumentException.class)
 	public void testNoStartStep() throws Exception {
-		flow.setStateTransitions(collect(StateTransition.createStateTransition(new StateSupport<String>("step"),
+		flow.setStateTransitions(collect(StateTransition.createStateTransition(new StateSupport<Object>("step"),
 				FlowExecution.FAILED, "step"), StateTransition
-				.createEndStateTransition(new StateSupport<String>("step"))));
+				.createEndStateTransition(new StateSupport<Object>("step"))));
 		flow.afterPropertiesSet();
 	}
 
 	@Test(expected = IllegalArgumentException.class)
 	public void testNoEndStep() throws Exception {
-		flow.setStateTransitions(Collections.singleton(StateTransition.createStateTransition(new StateSupport<String>(
+		flow.setStateTransitions(Collections.singleton(StateTransition.createStateTransition(new StateSupport<Object>(
 				"step"), FlowExecution.FAILED, "step")));
 		flow.setStartStateName("step");
 		flow.afterPropertiesSet();
@@ -101,6 +102,24 @@ public class BasicFlowTests {
 		assertEquals("step1", execution.getName());
 	}
 
+	@Test
+	public void testOneStepWithListenerCallsClose() throws Exception {
+		flow.setStateTransitions(Collections
+				.singleton(StateTransition.createEndStateTransition(new StubState("step1"))));
+		flow.afterPropertiesSet();
+		final List<FlowExecution> list = new ArrayList<FlowExecution>();
+		executor = new FlowExecutionListenerSupport() {
+			@Override
+			public void close(FlowExecution result) {
+				list.add(result);
+			}
+		};
+		FlowExecution execution = flow.start(executor);
+		assertEquals(1, list.size());
+		assertEquals(FlowExecution.COMPLETED, execution.getStatus());
+		assertEquals("step1", execution.getName());
+	}
+
 	@Test
 	public void testExplicitStartStep() throws Exception {
 		flow.setStateTransitions(collect(StateTransition.createStateTransition(new StubState("step"),
@@ -136,7 +155,7 @@ public class BasicFlowTests {
 	public void testFailedStep() throws Exception {
 		flow.setStateTransitions(collect(StateTransition.createStateTransition(new StubState("step1") {
 			@Override
-			public String handle(String executor) {
+			public String handle(Object executor) {
 				return FlowExecution.FAILED;
 			}
 		}, "step2"), StateTransition.createEndStateTransition(new StubState("step2"))));
@@ -165,7 +184,7 @@ public class BasicFlowTests {
 					private boolean paused = false;
 
 					@Override
-					public String handle(String executor) throws Exception {
+					public String handle(Object executor) throws Exception {
 						if (!paused) {
 							paused = true;
 							return FlowExecution.PAUSED;
@@ -184,23 +203,23 @@ public class BasicFlowTests {
 		assertEquals("step3", execution.getName());
 	}
 
-	private Collection<StateTransition<String>> collect(StateTransition<String> s1, StateTransition<String> s2) {
-		Collection<StateTransition<String>> list = new ArrayList<StateTransition<String>>();
+	private Collection<StateTransition<Object>> collect(StateTransition<Object> s1, StateTransition<Object> s2) {
+		Collection<StateTransition<Object>> list = new ArrayList<StateTransition<Object>>();
 		list.add(s1);
 		list.add(s2);
 		return list;
 	}
 
-	private Collection<StateTransition<String>> collect(StateTransition<String> s1, StateTransition<String> s2,
-			StateTransition<String> s3) {
-		Collection<StateTransition<String>> list = collect(s1, s2);
+	private Collection<StateTransition<Object>> collect(StateTransition<Object> s1, StateTransition<Object> s2,
+			StateTransition<Object> s3) {
+		Collection<StateTransition<Object>> list = collect(s1, s2);
 		list.add(s3);
 		return list;
 	}
 
-	private Collection<StateTransition<String>> collect(StateTransition<String> s1, StateTransition<String> s2,
-			StateTransition<String> s3, StateTransition<String> s4) {
-		Collection<StateTransition<String>> list = collect(s1, s2, s3);
+	private Collection<StateTransition<Object>> collect(StateTransition<Object> s1, StateTransition<Object> s2,
+			StateTransition<Object> s3, StateTransition<Object> s4) {
+		Collection<StateTransition<Object>> list = collect(s1, s2, s3);
 		list.add(s4);
 		return list;
 	}
@@ -209,7 +228,7 @@ public class BasicFlowTests {
 	 * @author Dave Syer
 	 * 
 	 */
-	private static class StubState extends StateSupport<String> {
+	private static class StubState extends StateSupport<Object> {
 
 		/**
 		 * @param string