diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java
index a44ab0bef..e19536ee2 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java
@@ -15,13 +15,24 @@
  */
 package org.springframework.batch.core.step;
 
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.Date;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.JobInterruptedException;
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.StepExecutionListener;
 import org.springframework.batch.core.UnexpectedJobExecutionException;
+import org.springframework.batch.core.launch.support.ExitCodeMapper;
 import org.springframework.batch.core.listener.CompositeStepExecutionListener;
 import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.repository.NoSuchJobException;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.util.Assert;
 
@@ -35,6 +46,13 @@ import org.springframework.util.Assert;
  */
 public abstract class AbstractStep implements Step, InitializingBean {
 
+	/**
+	 * Exit code for interrupted status.
+	 */
+	public static final String JOB_INTERRUPTED = "JOB_INTERRUPTED";
+
+	private static final Log logger = LogFactory.getLog(AbstractStep.class);
+
 	protected String name;
 
 	protected int startLimit = Integer.MAX_VALUE;
@@ -42,7 +60,7 @@ public abstract class AbstractStep implements Step, InitializingBean {
 	protected boolean allowStartIfComplete;
 
 	private CompositeStepExecutionListener listener = new CompositeStepExecutionListener();
-	
+
 	private JobRepository jobRepository;
 
 	/**
@@ -51,13 +69,11 @@ public abstract class AbstractStep implements Step, InitializingBean {
 	public AbstractStep() {
 		super();
 	}
-	
+
 	public void afterPropertiesSet() throws Exception {
 		Assert.notNull(jobRepository, "JobRepository is mandatory");
 	}
 
-
-
 	public String getName() {
 		return this.name;
 	}
@@ -107,8 +123,118 @@ public abstract class AbstractStep implements Step, InitializingBean {
 		this.name = name;
 	}
 
-	public abstract void execute(StepExecution stepExecution) throws JobInterruptedException,
-			UnexpectedJobExecutionException;
+	protected abstract ExitStatus doExecute(StepExecution stepExecution) throws Exception;
+
+	protected abstract void open(ExecutionContext ctx) throws Exception;
+
+	protected abstract void close(ExecutionContext ctx) throws Exception;
+
+	/**
+	 * Template method for step execution logic - calls abstract methods for
+	 * resource initialization ({@link #open(ExecutionContext)}), execution
+	 * logic ({@link #doExecute(StepExecution)}) and resource closing ({@link #close(ExecutionContext)}).
+	 */
+	public void execute(StepExecution stepExecution) throws JobInterruptedException, UnexpectedJobExecutionException {
+		stepExecution.setStartTime(new Date());
+		stepExecution.setStatus(BatchStatus.STARTED);
+
+		ExitStatus exitStatus = ExitStatus.FAILED;
+		Exception commitException = null;
+
+		try {
+			getCompositeListener().beforeStep(stepExecution);
+			try {
+				open(stepExecution.getExecutionContext());
+			}
+			catch (Exception e) {
+				throw new UnexpectedJobExecutionException("Failed to initialize the step", e);
+			}
+			exitStatus = doExecute(stepExecution);
+			exitStatus = exitStatus.and(getCompositeListener().afterStep(stepExecution));
+
+			try {
+				getJobRepository().saveOrUpdateExecutionContext(stepExecution);
+				stepExecution.setStatus(BatchStatus.COMPLETED);
+			}
+			catch (Exception e) {
+				commitException = e;
+				stepExecution.setStatus(BatchStatus.UNKNOWN);
+			}
+
+		}
+		catch (Throwable e) {
+
+			logger.error("Encountered an error executing the step");
+			stepExecution.setStatus(determineBatchStatus(e));
+			exitStatus = getDefaultExitStatusForFailure(e);
+
+			try {
+				exitStatus = exitStatus.and(getCompositeListener().onErrorInStep(stepExecution, e));
+			}
+			catch (Exception ex) {
+				logger.error("Encountered an error on listener close.", ex);
+			}
+			rethrow(e);
+		}
+		finally {
+
+			stepExecution.setExitStatus(exitStatus);
+			stepExecution.setEndTime(new Date());
+
+			try {
+				getJobRepository().saveOrUpdate(stepExecution);
+			}
+			catch (Exception e) {
+				commitException = e;
+			}
+
+			try {
+				close(stepExecution.getExecutionContext());
+			}
+			catch (Exception e) {
+				logger.error("Exception while closing step's resources", e);
+				throw new UnexpectedJobExecutionException("Exception while closing step's resources", e);
+			}
+
+			if (commitException != null) {
+				logger.error("Encountered an error saving batch meta data."
+						+ "This job is now in an unknown state and should not be restarted.", commitException);
+				throw new UnexpectedJobExecutionException("Encountered an error saving batch meta data.",
+						commitException);
+			}
+		}
+	}
+
+	private static void rethrow(Throwable e) throws JobInterruptedException {
+		if (e instanceof Error) {
+			throw (Error) e;
+		}
+		if (e instanceof JobInterruptedException) {
+			throw (JobInterruptedException) e;
+		}
+		else if (e.getCause() instanceof JobInterruptedException) {
+			throw (JobInterruptedException) e.getCause();
+		}
+		else if (e instanceof RuntimeException) {
+			throw (RuntimeException) e;
+		}
+		throw new UnexpectedJobExecutionException(e);
+	}
+
+	/**
+	 * Determine the step status based on the exception.
+	 */
+	private static BatchStatus determineBatchStatus(Throwable e) {
+		if (e instanceof FatalException) {
+			return BatchStatus.UNKNOWN;
+		}
+		else if (e instanceof JobInterruptedException || e.getCause() instanceof JobInterruptedException) {
+			return BatchStatus.STOPPED;
+		}
+		else {
+			return BatchStatus.FAILED;
+		}
+	}
 
 	/**
 	 * Register a step listener for callbacks at the appropriate stages in a
@@ -137,7 +263,7 @@ public abstract class AbstractStep implements Step, InitializingBean {
 	protected StepExecutionListener getCompositeListener() {
 		return listener;
 	}
-	
+
 	/**
 	 * Public setter for {@link JobRepository}.
 	 * 
@@ -150,7 +276,44 @@ public abstract class AbstractStep implements Step, InitializingBean {
 	protected JobRepository getJobRepository() {
 		return jobRepository;
 	}
-	
-	
+
+	/**
+	 * Default mapping from throwable to {@link ExitStatus}. Clients can modify
+	 * the exit code using a {@link StepExecutionListener}.
+	 * 
+	 * @param ex the cause of the failure
+	 * @return an {@link ExitStatus}
+	 */
+	private ExitStatus getDefaultExitStatusForFailure(Throwable ex) {
+		ExitStatus exitStatus;
+		if (ex instanceof JobInterruptedException || ex.getCause() instanceof JobInterruptedException) {
+			exitStatus = new ExitStatus(false, JOB_INTERRUPTED, JobInterruptedException.class.getName());
+		}
+		else if (ex instanceof NoSuchJobException || ex.getCause() instanceof NoSuchJobException) {
+			exitStatus = new ExitStatus(false, ExitCodeMapper.NO_SUCH_JOB);
+		}
+		else {
+			String message = "";
+			if (ex != null) {
+				StringWriter writer = new StringWriter();
+				ex.printStackTrace(new PrintWriter(writer));
+				message = writer.toString();
+			}
+			exitStatus = ExitStatus.FAILED.addExitDescription(message);
+		}
+
+		return exitStatus;
+	}
+
+	/**
+	 * Signals a fatal exception - e.g. unable to persist batch metadata or
+	 * rollback transaction. Throwing this exception will result in storing
+	 * {@link BatchStatus#UNKNOWN} as step's status.
+	 */
+	protected class FatalException extends RuntimeException {
+		public FatalException(String string, Exception e) {
+			super(string, e);
+		}
+	}
 
 }
\ No newline at end of file
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java
index 6108d215f..0eefe3f36 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java
@@ -15,26 +15,18 @@
  */
 package org.springframework.batch.core.step.item;
 
-import java.io.PrintWriter;
-import java.io.StringWriter;
-import java.util.Date;
-
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.batch.core.BatchStatus;
-import org.springframework.batch.core.JobInterruptedException;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.StepExecutionListener;
-import org.springframework.batch.core.UnexpectedJobExecutionException;
-import org.springframework.batch.core.launch.support.ExitCodeMapper;
-import org.springframework.batch.core.repository.JobRepository;
-import org.springframework.batch.core.repository.NoSuchJobException;
 import org.springframework.batch.core.step.AbstractStep;
 import org.springframework.batch.core.step.StepExecutionSynchronizer;
 import org.springframework.batch.core.step.StepExecutionSynchronizerFactory;
 import org.springframework.batch.core.step.StepInterruptionPolicy;
 import org.springframework.batch.core.step.ThreadStepInterruptionPolicy;
+import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemWriter;
@@ -73,11 +65,6 @@ public class ItemOrientedStep extends AbstractStep {
 
 	private static final Log logger = LogFactory.getLog(ItemOrientedStep.class);
 
-	/**
-	 * Exit code for interrupted status.
-	 */
-	public static final String JOB_INTERRUPTED = "JOB_INTERRUPTED";
-
 	private RepeatOperations chunkOperations = new RepeatTemplate();
 
 	private RepeatOperations stepOperations = new RepeatTemplate();
@@ -218,231 +205,98 @@ public class ItemOrientedStep extends AbstractStep {
 	 * @throws JobInterruptedException if the step or a chunk is interrupted
 	 * @throws RuntimeException if there is an exception during a chunk
 	 * execution
-	 * @see StepExecutor#execute(StepExecution)
+	 * 
 	 */
-	public void execute(final StepExecution stepExecution) throws UnexpectedJobExecutionException,
-			JobInterruptedException {
+	public ExitStatus doExecute(final StepExecution stepExecution) throws Exception {
+		stream.update(stepExecution.getExecutionContext());
+		getJobRepository().saveOrUpdateExecutionContext(stepExecution);
+		itemHandler.mark();
 
-		ExitStatus status = ExitStatus.FAILED;
 		final ExceptionHolder fatalException = new ExceptionHolder();
 
-		try {
+		return stepOperations.iterate(new RepeatCallback() {
 
-			stepExecution.setStartTime(new Date(System.currentTimeMillis()));
-			// We need to save the step execution right away, before we start
-			// using its ID. It would be better to make the creation atomic in
-			// the caller.
-			fatalException.setException(updateStatus(stepExecution, BatchStatus.STARTED));
+			public ExitStatus doInIteration(RepeatContext context) throws Exception {
+				final StepContribution contribution = stepExecution.createStepContribution();
+				// Before starting a new transaction, check for
+				// interruption.
+				if (stepExecution.isTerminateOnly()) {
+					context.setTerminateOnly();
+				}
+				interruptionPolicy.checkInterrupted(stepExecution);
 
-			// Execute step level listeners *after* the execution context is
-			// fixed in the step. E.g. ItemStream instances need the the same
-			// reference to the ExecutionContext as the step execution.
-			getCompositeListener().beforeStep(stepExecution);
-			stream.open(stepExecution.getExecutionContext());
-			stream.update(stepExecution.getExecutionContext());
-			getJobRepository().saveOrUpdateExecutionContext(stepExecution);
-			itemHandler.mark();
+				ExitStatus exitStatus = ExitStatus.CONTINUABLE;
 
-			status = stepOperations.iterate(new RepeatCallback() {
+				TransactionStatus transaction = transactionManager.getTransaction(new DefaultTransactionDefinition());
 
-				public ExitStatus doInIteration(final RepeatContext context) throws Exception {
+				try {
 
-					final StepContribution contribution = stepExecution.createStepContribution();
-					// Before starting a new transaction, check for
-					// interruption.
-					if (stepExecution.isTerminateOnly()) {
-						context.setTerminateOnly();
-					}
-					interruptionPolicy.checkInterrupted(stepExecution);
-
-					ExitStatus result = ExitStatus.CONTINUABLE;
-
-					TransactionStatus transaction = transactionManager
-							.getTransaction(new DefaultTransactionDefinition());
+					exitStatus = processChunk(stepExecution, contribution);
+					contribution.incrementCommitCount();
 
+					// If the step operations are asynchronous then we need
+					// to synchronize changes to the step execution (at a
+					// minimum).
 					try {
-
-						result = processChunk(stepExecution, contribution);
-						contribution.incrementCommitCount();
-
-						// If the step operations are asynchronous then we need
-						// to synchronize changes to the step execution (at a
-						// minimum).
-						try {
-							synchronizer.lock(stepExecution);
-						}
-						catch (InterruptedException e) {
-							stepExecution.setStatus(BatchStatus.STOPPED);
-							Thread.currentThread().interrupt();
-						}
-
-						// Apply the contribution to the step
-						// only if chunk was successful
-						stepExecution.apply(contribution);
-
-						// Attempt to flush before the step execution and stream
-						// state are updated
-						itemHandler.flush();
-
-						stream.update(stepExecution.getExecutionContext());
-						try {
-							getJobRepository().saveOrUpdateExecutionContext(stepExecution);
-						}
-						catch (Exception e) {
-							fatalException.setException(e);
-							stepExecution.setStatus(BatchStatus.UNKNOWN);
-							throw new CommitFailedException(
-									"Fatal error detected during save of step execution context", e);
-						}
-
-						try {
-							itemHandler.mark();
-							transactionManager.commit(transaction);
-						}
-						catch (Exception e) {
-							fatalException.setException(e);
-							stepExecution.setStatus(BatchStatus.UNKNOWN);
-							throw new CommitFailedException("Fatal error detected during commit", e);
-						}
-
+						synchronizer.lock(stepExecution);
 					}
-					catch (Error e) {
-						processRollback(stepExecution, contribution, fatalException, transaction);
-						throw e;
+					catch (InterruptedException e) {
+						stepExecution.setStatus(BatchStatus.STOPPED);
+						Thread.currentThread().interrupt();
+					}
+
+					// Apply the contribution to the step
+					// only if chunk was successful
+					stepExecution.apply(contribution);
+
+					// Attempt to flush before the step execution and stream
+					// state are updated
+					itemHandler.flush();
+
+					stream.update(stepExecution.getExecutionContext());
+					try {
+						getJobRepository().saveOrUpdateExecutionContext(stepExecution);
 					}
 					catch (Exception e) {
-						processRollback(stepExecution, contribution, fatalException, transaction);
-						throw e;
-					}
-					finally {
-						synchronizer.release(stepExecution);
+						fatalException.setException(e);
+						stepExecution.setStatus(BatchStatus.UNKNOWN);
+						throw new FatalException("Fatal error detected during save of step execution context", e);
 					}
 
-					// Check for interruption after transaction as well, so that
-					// the interrupted exception is correctly propagated up to
-					// caller
-					interruptionPolicy.checkInterrupted(stepExecution);
-
-					return result;
+					try {
+						itemHandler.mark();
+						transactionManager.commit(transaction);
+					}
+					catch (Exception e) {
+						fatalException.setException(e);
+						stepExecution.setStatus(BatchStatus.UNKNOWN);
+						logger.error("Fatal error detected during commit.");
+						throw new FatalException("Fatal error detected during commit", e);
+					}
 
 				}
-			});
-
-			status = status.and(getCompositeListener().afterStep(stepExecution));
-
-			fatalException.setException(updateStatus(stepExecution, BatchStatus.COMPLETED));
-		}
-		catch (CommitFailedException e) {
-			logger.error("Fatal error detected during commit.");
-			throw e;
-		}
-		catch (RuntimeException e) {
-			status = processFailure(stepExecution, fatalException, e);
-			if (e.getCause() instanceof JobInterruptedException) {
-				updateStatus(stepExecution, BatchStatus.STOPPED);
-				throw (JobInterruptedException) e.getCause();
-			}
-			throw e;
-		}
-		catch (Error e) {
-			status = processFailure(stepExecution, fatalException, e);
-			throw e;
-		}
-		finally {
-
-			stepExecution.setExitStatus(status);
-			stepExecution.setEndTime(new Date(System.currentTimeMillis()));
-
-			try {
-				getJobRepository().saveOrUpdate(stepExecution);
-			}
-			catch (RuntimeException e) {
-				String msg = "Fatal error detected during final save of meta data";
-				logger.error(msg, e);
-				if (!fatalException.hasException()) {
-					fatalException.setException(e);
+				catch (Error e) {
+					processRollback(stepExecution, contribution, fatalException, transaction);
+					throw e;
 				}
-				throw new UnexpectedJobExecutionException(msg, fatalException.getException());
-			}
-
-			try {
-				stream.close(stepExecution.getExecutionContext());
-			}
-			catch (RuntimeException e) {
-				String msg = "Fatal error detected during close of streams. "
-						+ "The job execution completed (possibly unsuccessfully but with consistent meta-data).";
-				logger.error(msg, e);
-				if (!fatalException.hasException()) {
-					fatalException.setException(e);
+				catch (Exception e) {
+					processRollback(stepExecution, contribution, fatalException, transaction);
+					throw e;
 				}
-				throw new UnexpectedJobExecutionException(msg, fatalException.getException());
+				finally {
+					synchronizer.release(stepExecution);
+				}
+
+				// Check for interruption after transaction as well, so that
+				// the interrupted exception is correctly propagated up to
+				// caller
+				interruptionPolicy.checkInterrupted(stepExecution);
+
+				return exitStatus;
 			}
 
-			if (fatalException.hasException()) {
-				throw new UnexpectedJobExecutionException("Encountered an error saving batch meta data.",
-						fatalException.getException());
-			}
+		});
 
-		}
-
-	}
-
-	/**
-	 * @param stepExecution the current {@link StepExecution}
-	 * @param fatalException the {@link ExceptionHolder} containing information
-	 * about failures in meta-data
-	 * @param e the cause of teh failure
-	 * @return an {@link ExitStatus}
-	 */
-	private ExitStatus processFailure(final StepExecution stepExecution, final ExceptionHolder fatalException,
-			Throwable e) {
-
-		// Default classification marks this as a failure and adds the exception
-		// type and message
-		ExitStatus status = getDefaultExitStatusForFailure(e);
-
-		if (!fatalException.hasException()) {
-			try {
-				// classify exception so an exit code can be stored.
-				status = status.and(getCompositeListener().onErrorInStep(stepExecution, e));
-			}
-			catch (RuntimeException ex) {
-				logger.error("Unexpected error in listener on error in step.", ex);
-			}
-			updateStatus(stepExecution, BatchStatus.FAILED);
-		}
-		else {
-			logger.error("Fatal error detected during rollback caused by underlying exception: ", e);
-		}
-		return status;
-	}
-
-	/**
-	 * Default mapping from throwable to {@link ExitStatus}. Clients can modify
-	 * the exit code using a {@link StepExecutionListener}.
-	 * 
-	 * @param throwable the cause of teh failure
-	 * @return an {@link ExitStatus}
-	 */
-	private ExitStatus getDefaultExitStatusForFailure(Throwable throwable) {
-		ExitStatus exitStatus;
-		if (throwable instanceof JobInterruptedException) {
-			exitStatus = new ExitStatus(false, JOB_INTERRUPTED, JobInterruptedException.class.getName());
-		}
-		else if (throwable instanceof NoSuchJobException) {
-			exitStatus = new ExitStatus(false, ExitCodeMapper.NO_SUCH_JOB);
-		}
-		else {
-			String message = "";
-			if (throwable != null) {
-				StringWriter writer = new StringWriter();
-				throwable.printStackTrace(new PrintWriter(writer));
-				message = writer.toString();
-			}
-			exitStatus = ExitStatus.FAILED.addExitDescription(message);
-		}
-
-		return exitStatus;
 	}
 
 	/**
@@ -475,25 +329,6 @@ public class ItemOrientedStep extends AbstractStep {
 		return result;
 	}
 
-	/**
-	 * Convenience method to update the status in all relevant places.
-	 * 
-	 * @param stepInstance the current step
-	 * @param stepExecution the current stepExecution
-	 * @param status the status to set
-	 */
-	private Exception updateStatus(StepExecution stepExecution, BatchStatus status) {
-		stepExecution.setStatus(status);
-		try {
-			getJobRepository().saveOrUpdate(stepExecution);
-			return null;
-		}
-		catch (Exception e) {
-			return e;
-		}
-
-	}
-
 	/**
 	 * @param stepExecution
 	 * @param contribution
@@ -522,7 +357,7 @@ public class ItemOrientedStep extends AbstractStep {
 			 */
 			if (!fatalException.hasException()) {
 				fatalException.setException(e);
-				stepExecution.setStatus(BatchStatus.UNKNOWN);
+				throw new FatalException("Failed while processing rollback", e);
 			}
 		}
 	}
@@ -545,11 +380,11 @@ public class ItemOrientedStep extends AbstractStep {
 
 	}
 
-	private class CommitFailedException extends RuntimeException {
-
-		public CommitFailedException(String string, Exception e) {
-			super(string, e);
-		}
+	protected void close(ExecutionContext ctx) throws Exception {
+		stream.close(ctx);
+	}
 
+	protected void open(ExecutionContext ctx) throws Exception {
+		stream.open(ctx);
 	}
 }
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java
index 8ce2ed3d4..90744f836 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java
@@ -15,21 +15,14 @@
  */
 package org.springframework.batch.core.step.tasklet;
 
-import java.util.Date;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.springframework.batch.core.BatchStatus;
-import org.springframework.batch.core.UnexpectedJobExecutionException;
-import org.springframework.batch.core.JobInterruptedException;
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.StepExecutionListener;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.step.AbstractStep;
+import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.beans.factory.BeanNameAware;
-import org.springframework.beans.factory.InitializingBean;
 import org.springframework.util.Assert;
 
 /**
@@ -45,9 +38,7 @@ import org.springframework.util.Assert;
  * @author Ben Hale
  * @author Robert Kasanicky
  */
-public class TaskletStep extends AbstractStep implements Step, InitializingBean, BeanNameAware {
-
-	private static final Log logger = LogFactory.getLog(TaskletStep.class);
+public class TaskletStep extends AbstractStep implements BeanNameAware {
 
 	private Tasklet tasklet;
 
@@ -116,62 +107,21 @@ public class TaskletStep extends AbstractStep implements Step, InitializingBean,
 		this.tasklet = tasklet;
 	}
 
-	public void execute(StepExecution stepExecution) throws JobInterruptedException, UnexpectedJobExecutionException {
-		stepExecution.setStartTime(new Date());
-		stepExecution.setStatus(BatchStatus.STARTED);
-
-		ExitStatus exitStatus = ExitStatus.FAILED;
-		Exception fatalException = null;
-		try {
-
-			getCompositeListener().beforeStep(stepExecution);
-			exitStatus = tasklet.execute();
-			exitStatus = exitStatus.and(getCompositeListener().afterStep(stepExecution));
-
-			try {
-				getJobRepository().saveOrUpdateExecutionContext(stepExecution);
-				stepExecution.setStatus(BatchStatus.COMPLETED);
-			}
-			catch (Exception e) {
-				fatalException = e;
-				stepExecution.setStatus(BatchStatus.UNKNOWN);
-			}
-
-		}
-		catch (Exception e) {
-			logger.error("Encountered an error running the tasklet");
-			stepExecution.setStatus(BatchStatus.FAILED);
-			try {
-				exitStatus = exitStatus.and(getCompositeListener().onErrorInStep(stepExecution, e));
-			}
-			catch (Exception ex) {
-				logger.error("Encountered an error on listener close.", ex);
-			}
-			if (e instanceof JobInterruptedException) {
-				throw (JobInterruptedException) e;
-			}
-			else if (e instanceof RuntimeException) {
-				throw (RuntimeException) e;
-			}
-			throw new UnexpectedJobExecutionException(e);
-		}
-		finally {
-			stepExecution.setExitStatus(exitStatus);
-			stepExecution.setEndTime(new Date());
-			try {
-				getJobRepository().saveOrUpdate(stepExecution);
-			}
-			catch (Exception e) {
-				fatalException = e;
-			}
-			if (fatalException != null) {
-				logger.error("Encountered an error saving batch meta data."
-						+ "This job is now in an unknown state and should not be restarted.", fatalException);
-				throw new UnexpectedJobExecutionException("Encountered an error saving batch meta data.",
-						fatalException);
-			}
-		}
-
+	/**
+	 * Delegate to tasklet.
+	 */
+	protected ExitStatus doExecute(StepExecution stepExecution) throws Exception {
+		return tasklet.execute();
 	}
 
+	protected void close(ExecutionContext ctx) throws Exception {
+	}
+
+	protected void open(ExecutionContext ctx) throws Exception {
+	}
+
+	
+
+	
+
 }
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java
index 954560a7f..0d01946cd 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java
@@ -45,6 +45,7 @@ import org.springframework.batch.core.step.AbstractStep;
 import org.springframework.batch.core.step.skip.ItemSkipPolicy;
 import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy;
 import org.springframework.batch.item.AbstractItemReader;
+import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.repeat.ExitStatus;
@@ -510,5 +511,20 @@ public class SimpleJobTests extends TestCase {
 			this.itemSkipPolicy = itemSkipPolicy;
 		}
 
+		protected ExitStatus doExecute(StepExecution stepExecution) throws Exception {
+			// TODO Auto-generated method stub
+			return null;
+		}
+
+		protected void close(ExecutionContext ctx) throws Exception {
+			// TODO Auto-generated method stub
+			
+		}
+
+		protected void open(ExecutionContext ctx) throws Exception {
+			// TODO Auto-generated method stub
+			
+		}
+
 	}
 }
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepTests.java
index 5bdba71b8..1a5041755 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepTests.java
@@ -279,8 +279,10 @@ public class ItemOrientedStepTests extends TestCase {
 		});
 		itemOrientedStep.execute(stepExecution);
 
-		// context saved before processing starts and updated at the end
-		assertEquals(2, list.size());
+		// context saved before looping and updated once for every processing
+		// loop (once in this case) and finally in the abstract step (regardless
+		// of execution logic)
+		assertEquals(3, list.size());
 	}
 
 	public void testSuccessfulExecutionWithFailureOnSaveOfExecutionContext() throws Exception {
@@ -298,37 +300,16 @@ public class ItemOrientedStepTests extends TestCase {
 		});
 		try {
 			itemOrientedStep.execute(stepExecution);
-			fail("Expected BatchCriticalException");
+			fail();
 		}
-		catch (UnexpectedJobExecutionException e) {
+		catch (RuntimeException e) {
+			assertEquals("Fatal error detected during save of step execution context", e.getMessage());
 			assertEquals("foo", e.getCause().getMessage());
 		}
 		assertEquals(BatchStatus.UNKNOWN, stepExecution.getStatus());
 	}
 
-	/*
-	 * make sure a job that has been executed before, and is therefore being
-	 * restarted, is restored.
-	 */
-	// public void testRestartedJob() throws Exception {
-	// String step = "stepName";
-	// // step.setStepExecutionCount(1);
-	// MockRestartableItemReader tasklet = new MockRestartableItemReader();
-	// stepExecutor.setItemReader(tasklet);
-	// stepConfiguration.setSaveExecutionContext(true);
-	// JobExecution jobExecution = new JobExecution(jobInstance);
-	// StepExecution stepExecution = new StepExecution(step, jobExecution);
-	//
-	// stepExecution
-	// .setExecutionContext(new
-	// ExecutionContext(PropertiesConverter.stringToProperties("foo=bar")));
-	// // step.setLastExecution(stepExecution);
-	// stepExecutor.execute(stepExecution);
-	//
-	// assertTrue(tasklet.isRestoreFromCalled());
-	// assertTrue(tasklet.isRestoreFromCalledWithSomeContext());
-	// assertTrue(tasklet.isGetExecutionAttributesCalled());
-	// }
+
 	/*
 	 * Test that a job that is being restarted, but has saveExecutionAttributes
 	 * set to false, doesn't have restore or getExecutionAttributes called on
@@ -497,7 +478,6 @@ public class ItemOrientedStepTests extends TestCase {
 			}
 
 			public void update(ExecutionContext executionContext) {
-				// TODO Auto-generated method stub
 				executionContext.putString("foo", "bar");
 			}
 		};
@@ -547,7 +527,6 @@ public class ItemOrientedStepTests extends TestCase {
 		StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecutionContext);
 
 		stepExecution.setExecutionContext(new ExecutionContext(PropertiesConverter.stringToProperties("foo=bar")));
-		// step.setLastExecution(stepExecution);
 
 		try {
 			itemOrientedStep.execute(stepExecution);
@@ -556,7 +535,7 @@ public class ItemOrientedStepTests extends TestCase {
 		catch (JobInterruptedException ex) {
 			assertEquals(BatchStatus.STOPPED, stepExecution.getStatus());
 			String msg = stepExecution.getExitStatus().getExitDescription();
-			assertTrue("Message does not contain 'interrupted': " + msg, contains(msg, "interrupted"));
+			assertTrue("Message does not contain 'JobInterruptedException': " + msg, contains(msg, "JobInterruptedException"));
 		}
 	}
 
@@ -640,7 +619,7 @@ public class ItemOrientedStepTests extends TestCase {
 			itemOrientedStep.execute(stepExecution);
 			fail("Expected UnexpectedJobExecutionException");
 		}
-		catch (UnexpectedJobExecutionException ex) {
+		catch (RuntimeException ex) {
 			assertEquals(BatchStatus.UNKNOWN, stepExecution.getStatus());
 			String msg = stepExecution.getExitStatus().getExitDescription();
 			assertTrue("Message does not contain ResetFailedException: " + msg, contains(msg, "ResetFailedException"));
@@ -668,12 +647,12 @@ public class ItemOrientedStepTests extends TestCase {
 			itemOrientedStep.execute(stepExecution);
 			fail("Expected BatchCriticalException");
 		}
-		catch (UnexpectedJobExecutionException ex) {
+		catch (RuntimeException ex) {
 			assertEquals(BatchStatus.UNKNOWN, stepExecution.getStatus());
 			String msg = stepExecution.getExitStatus().getExitDescription();
-			assertEquals("", msg);
+			assertTrue(msg.contains("Fatal error detected during commit"));
 			msg = ex.getMessage();
-			assertTrue("Message does not contain 'saving': " + msg, contains(msg, "saving"));
+			assertTrue(msg.contains("Fatal error detected during commit"));
 			// The original rollback was caused by this one:
 			assertEquals("Bar", ex.getCause().getMessage());
 		}
@@ -703,7 +682,7 @@ public class ItemOrientedStepTests extends TestCase {
 			String msg = stepExecution.getExitStatus().getExitDescription();
 			assertEquals("", msg);
 			msg = ex.getMessage();
-			assertTrue("Message does not contain 'final': " + msg, contains(msg, "final"));
+			assertTrue("Message does not contain 'saving batch meta data': " + msg, contains(msg, "saving batch meta data"));
 			// The original rollback was caused by this one:
 			assertEquals("Bar", ex.getCause().getMessage());
 		}
@@ -737,7 +716,7 @@ public class ItemOrientedStepTests extends TestCase {
 			String msg = stepExecution.getExitStatus().getExitDescription();
 			assertEquals("", msg);
 			msg = ex.getMessage();
-			assertTrue("Message does not contain 'close': " + msg, contains(msg, "close"));
+			assertTrue("Message does not contain 'closing': " + msg, contains(msg, "closing"));
 			// The original rollback was caused by this one:
 			assertEquals("Bar", ex.getCause().getMessage());
 		}
@@ -792,9 +771,9 @@ public class ItemOrientedStepTests extends TestCase {
 		catch (RuntimeException expected) {
 			assertEquals("exception thrown in afterStep to signal failure", expected.getMessage());
 		}
-		
+
 		assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
-		
+
 	}
 
 	private boolean contains(String str, String searchStr) {
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java
index 8133f22a4..c7a33d495 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java
@@ -130,7 +130,7 @@ public class TaskletStepTests extends TestCase {
 		}
 		catch (RuntimeException e) {
 			assertNotNull(stepExecution.getStartTime());
-			assertEquals(ExitStatus.FAILED, stepExecution.getExitStatus());
+			assertEquals(ExitStatus.FAILED.getExitCode(), stepExecution.getExitStatus().getExitCode());
 			assertNotNull(stepExecution.getEndTime());
 		}
 	}
@@ -143,7 +143,7 @@ public class TaskletStepTests extends TestCase {
 		}
 		catch (Error e) {
 			assertNotNull(stepExecution.getStartTime());
-			assertEquals(ExitStatus.FAILED, stepExecution.getExitStatus());
+			assertEquals(ExitStatus.FAILED.getExitCode(), stepExecution.getExitStatus().getExitCode());
 			assertNotNull(stepExecution.getEndTime());
 		}
 	}