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 d62e71904..fdf90a2cf 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
@@ -272,10 +272,7 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
 	 * 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) {
+		if (e instanceof JobInterruptedException || e.getCause() instanceof JobInterruptedException) {
 			return BatchStatus.STOPPED;
 		}
 		else {
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/FatalException.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/FatalException.java
deleted file mode 100644
index 11b299b86..000000000
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/FatalException.java
+++ /dev/null
@@ -1,20 +0,0 @@
-package org.springframework.batch.core.step;
-
-import org.springframework.batch.core.BatchStatus;
-import org.springframework.batch.core.UnexpectedJobExecutionException;
-
-/**
- * Signals a fatal exception in step - e.g. unable to persist batch metadata or
- * rollback transaction. Throwing this exception in a step implementation will
- * result in the step having a status of {@link BatchStatus#UNKNOWN}.
- */
-public class FatalException extends UnexpectedJobExecutionException {
-
-	public FatalException(String string, Throwable e) {
-		super(string, e);
-	}
-
-	public FatalException(String string) {
-		super(string);
-	}
-}
\ No newline at end of file
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java
index 8dc650afb..061b328a3 100755
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java
@@ -28,7 +28,6 @@ import org.springframework.batch.classify.Classifier;
 import org.springframework.batch.classify.SubclassClassifier;
 import org.springframework.batch.core.JobInterruptedException;
 import org.springframework.batch.core.Step;
-import org.springframework.batch.core.step.FatalException;
 import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
 import org.springframework.batch.core.step.skip.NonSkippableReadException;
 import org.springframework.batch.core.step.skip.SkipLimitExceededException;
@@ -294,9 +293,9 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 	@SuppressWarnings("unchecked")
 	@Override
 	protected void applyConfiguration(TaskletStep step) {
-		addFatalExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class, FatalException.class,
+		addFatalExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
 				SkipListenerFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
-		addNonRetryableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class, FatalException.class,
+		addNonRetryableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
 				SkipListenerFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
 
 		super.applyConfiguration(step);
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 5aae55587..ff76e0c42 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
@@ -30,7 +30,6 @@ import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.scope.context.ChunkContext;
 import org.springframework.batch.core.scope.context.StepContextRepeatCallback;
 import org.springframework.batch.core.step.AbstractStep;
-import org.springframework.batch.core.step.FatalException;
 import org.springframework.batch.core.step.StepInterruptionPolicy;
 import org.springframework.batch.core.step.ThreadStepInterruptionPolicy;
 import org.springframework.batch.item.ExecutionContext;
@@ -289,6 +288,15 @@ public class TaskletStep extends AbstractStep {
 		stream.open(ctx);
 	}
 
+	/**
+	 * A callback for the transactional work inside a chunk. Also detects
+	 * failures in the transaction commit and rollback, only panicking if the
+	 * transaction status is unknown (i.e. if a commit failure leads to a clean
+	 * rollback then we assume the state is consistent).
+	 * 
+	 * @author Dave Syer
+	 * 
+	 */
 	private class ChunkTransactionCallback extends TransactionSynchronizationAdapter implements TransactionCallback {
 
 		private final StepExecution stepExecution;
@@ -311,11 +319,12 @@ public class TaskletStep extends AbstractStep {
 					// Wah! the commit failed. We need to rescue the step
 					// execution data.
 					stepExecution.setVersion(oldVersion);
-				}				
+				}
 			}
 			if (status == TransactionSynchronization.STATUS_UNKNOWN) {
 				rollback(stepExecution);
 				stepExecution.upgradeStatus(BatchStatus.UNKNOWN);
+				stepExecution.setTerminateOnly();
 			}
 		}
 
@@ -359,6 +368,7 @@ public class TaskletStep extends AbstractStep {
 					}
 					catch (InterruptedException e) {
 						stepExecution.setStatus(BatchStatus.STOPPED);
+						stepExecution.setTerminateOnly();
 						Thread.currentThread().interrupt();
 					}
 
@@ -380,18 +390,15 @@ public class TaskletStep extends AbstractStep {
 					// stay false and we can use that later.
 					getJobRepository().updateExecutionContext(stepExecution);
 					stepExecution.incrementCommitCount();
-					/*
-					 * The step execution has to be saved before commit because
-					 * otherwise there is a deadlock between the data source
-					 * pool and the semaphore. As long as only one connection is
-					 * used inside the section of this callback that is locked
-					 * with the semaphore, the deadlock is avoided.
-					 */
 					logger.debug("Saving step execution before commit: " + stepExecution);
 					getJobRepository().update(stepExecution);
 				}
 				catch (Exception e) {
-					throw new FatalException("Fatal failure detected", e);
+					// If we get to here there was a problem saving the step
+					// execution and we have to fail.
+					stepExecution.upgradeStatus(BatchStatus.UNKNOWN);
+					stepExecution.setTerminateOnly();
+					throw e;
 				}
 
 			}
@@ -433,6 +440,13 @@ public class TaskletStep extends AbstractStep {
 
 	}
 
+	/**
+	 * Convenience wrapper for a checked exception so that it can cause a
+	 * rollback and be extracted afterwards.
+	 * 
+	 * @author Dave Syer
+	 * 
+	 */
 	private static class TransactionException extends RuntimeException {
 
 		public TransactionException(Exception e) {
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/ChunkElementParserTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/ChunkElementParserTests.java
index 02307c44a..894711432 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/ChunkElementParserTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/ChunkElementParserTests.java
@@ -50,7 +50,7 @@ public class ChunkElementParserTests {
 	public void testInheritSkippable() throws Exception {
 		Map<Class<? extends Throwable>, Boolean> skippable = getExceptionClasses("s1",
 				chunkElementParentAttributeParserTestsContext);
-		assertEquals(12, skippable.size());
+		assertEquals(11, skippable.size());
 		containsClassified(skippable, NullPointerException.class, true);
 		containsClassified(skippable, ArithmeticException.class, true);
 		containsClassified(skippable, CannotAcquireLockException.class, false);
@@ -61,7 +61,7 @@ public class ChunkElementParserTests {
 	public void testInheritSkippableWithNoMerge() throws Exception {
 		Map<Class<? extends Throwable>, Boolean> skippable = getExceptionClasses("s2",
 				chunkElementParentAttributeParserTestsContext);
-		assertEquals(10, skippable.size());
+		assertEquals(9, skippable.size());
 		containsClassified(skippable, NullPointerException.class, true);
 		assertFalse(skippable.containsKey(ArithmeticException.class));
 		containsClassified(skippable, CannotAcquireLockException.class, false);
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantExceptionClassesTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantExceptionClassesTests.java
index 8c3bcde5c..cd856f4ab 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantExceptionClassesTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantExceptionClassesTests.java
@@ -21,7 +21,6 @@ import org.springframework.batch.core.repository.JobExecutionAlreadyRunningExcep
 import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.JobRestartException;
-import org.springframework.batch.core.step.FatalException;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
@@ -122,15 +121,6 @@ public class FaultTolerantExceptionClassesTests implements ApplicationContextAwa
 		assertEquals("[]", writer.getCommitted().toString());
 	}
 
-	@Test
-	public void testInternalFatalChecked() throws Exception {
-		writer.setExceptionType(FatalException.class);
-		StepExecution stepExecution = launchStep("skippableFatalStep");
-		assertEquals(BatchStatus.UNKNOWN, stepExecution.getStatus());
-		assertEquals("[1, 2, 3]", writer.getWritten().toString());
-		assertEquals("[]", writer.getCommitted().toString());
-	}
-
 	@Test
 	public void testSkippableChecked() throws Exception {
 		writer.setExceptionType(SkippableException.class);
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 91689a8e1..ff1e0cafd 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
@@ -371,8 +371,7 @@ public class TaskletStepTests {
 
 		step.execute(stepExecution);
 		Throwable e = stepExecution.getFailureExceptions().get(0);
-		assertEquals("Fatal failure detected", e.getMessage());
-		assertEquals("foo", e.getCause().getMessage());
+		assertEquals("foo", e.getMessage());
 		assertEquals(BatchStatus.UNKNOWN, stepExecution.getStatus());
 	}