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 b0ecb8ed2..fb3525810 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
@@ -184,7 +184,6 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
 
 		// Start with a default value that will be trumped by anything
 		ExitStatus exitStatus = ExitStatus.EXECUTING;
-		Exception commitException = null;
 
 		StepSynchronizationManager.register(stepExecution);
 
@@ -234,8 +233,8 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
 				stepExecution.setStatus(BatchStatus.UNKNOWN);
 				exitStatus = exitStatus.and(ExitStatus.UNKNOWN);
 				stepExecution.addFailureException(e);
-				logger.error("Encountered an error saving batch meta data."
-						+ "This job is now in an unknown state and should not be restarted.", commitException);
+				logger.error("Encountered an error saving batch meta data. "
+						+ "This job is now in an unknown state and should not be restarted.", e);
 			}
 
 			stepExecution.setEndTime(new Date());
@@ -248,8 +247,8 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
 				stepExecution.setStatus(BatchStatus.UNKNOWN);
 				stepExecution.setExitStatus(exitStatus.and(ExitStatus.UNKNOWN));
 				stepExecution.addFailureException(e);
-				logger.error("Encountered an error saving batch meta data."
-						+ "This job is now in an unknown state and should not be restarted.", commitException);
+				logger.error("Encountered an error saving batch meta data. "
+						+ "This job is now in an unknown state and should not be restarted.", e);
 			}
 
 			try {
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 c263f8587..698084c7c 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
@@ -53,6 +53,7 @@ import org.springframework.batch.retry.policy.RetryContextCache;
 import org.springframework.batch.retry.policy.SimpleRetryPolicy;
 import org.springframework.core.task.SyncTaskExecutor;
 import org.springframework.core.task.TaskExecutor;
+import org.springframework.transaction.TransactionException;
 import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
 import org.springframework.transaction.interceptor.TransactionAttribute;
 
@@ -323,7 +324,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 	protected void applyConfiguration(TaskletStep step) {
 		addNonSkippableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
 				SkipListenerFailedException.class, SkipPolicyFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
-		addNonRetryableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
+		addNonRetryableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class, TransactionException.class,
 				SkipListenerFailedException.class, SkipPolicyFailedException.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 974ca3000..20eb52849 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
@@ -261,7 +261,7 @@ public class TaskletStep extends AbstractStep {
 					result = (RepeatStatus) new TransactionTemplate(transactionManager, transactionAttribute)
 							.execute(new ChunkTransactionCallback(chunkContext));
 				}
-				catch (TransactionException e) {
+				catch (UncheckedTransactionException e) {
 					// Allow checked exceptions to be thrown inside callback
 					throw (Exception) e.getCause();
 				}
@@ -430,7 +430,7 @@ public class TaskletStep extends AbstractStep {
 				logger.debug("Rollback for Exception: " + e.getClass().getName() + ": " + e.getMessage());
 				rollback(stepExecution);
 				// Allow checked exceptions
-				throw new TransactionException(e);
+				throw new UncheckedTransactionException(e);
 			}
 
 			return result;
@@ -453,9 +453,9 @@ public class TaskletStep extends AbstractStep {
 	 * @author Dave Syer
 	 * 
 	 */
-	private static class TransactionException extends RuntimeException {
+	private static class UncheckedTransactionException extends RuntimeException {
 
-		public TransactionException(Exception e) {
+		public UncheckedTransactionException(Exception e) {
 			super(e);
 		}
 
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java
index ce3a78e67..487d66225 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java
@@ -23,10 +23,13 @@ import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
 import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.transaction.TransactionException;
+import org.springframework.transaction.UnexpectedRollbackException;
 import org.springframework.transaction.interceptor.RollbackRuleAttribute;
 import org.springframework.transaction.interceptor.RuleBasedTransactionAttribute;
 import org.springframework.transaction.interceptor.TransactionAttribute;
 import org.springframework.transaction.interceptor.TransactionAttributeEditor;
+import org.springframework.transaction.support.DefaultTransactionStatus;
 import org.springframework.util.StringUtils;
 
 /**
@@ -62,7 +65,8 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
 		factory = new FaultTolerantStepFactoryBean<String, String>();
 
 		factory.setBeanName("stepName");
-		factory.setTransactionManager(new ResourcelessTransactionManager());
+		ResourcelessTransactionManager transactionManager = new ResourcelessTransactionManager();
+		factory.setTransactionManager(transactionManager);
 		factory.setCommitInterval(2);
 
 		reader.clear();
@@ -78,6 +82,7 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
 		factory.setSkippableExceptionClasses(getExceptionMap(Exception.class));
 
 		MapJobRepositoryFactoryBean repositoryFactory = new MapJobRepositoryFactoryBean();
+		repositoryFactory.setTransactionManager(transactionManager);
 		repositoryFactory.afterPropertiesSet();
 		repository = (JobRepository) repositoryFactory.getObject();
 		factory.setJobRepository(repository);
@@ -468,6 +473,40 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
 		assertEquals("[1, 2, 3, 4, 5]", processor.getProcessed().toString());
 	}
 
+	@Test
+	public void testTransactionException() throws Exception {
+		ResourcelessTransactionManager transactionManager = new ResourcelessTransactionManager() {
+			private boolean failed = false;
+			protected void doCommit(DefaultTransactionStatus status) throws TransactionException {
+				if (writer.getWritten().isEmpty() || failed || !status.isNewTransaction()) {
+					super.doCommit(status);
+					return;
+				}
+				failed = true;
+				status.setRollbackOnly();
+				super.doRollback(status);
+				throw new UnexpectedRollbackException("Planned");
+			}
+		};
+		MapJobRepositoryFactoryBean repositoryFactory = new MapJobRepositoryFactoryBean();
+		repositoryFactory.setTransactionManager(transactionManager);
+		repositoryFactory.afterPropertiesSet();
+		repository = (JobRepository) repositoryFactory.getObject();
+		factory.setJobRepository(repository);
+		factory.setTransactionManager(transactionManager);
+
+		jobExecution = repository.createJobExecution("skipJob", new JobParameters());
+		stepExecution = jobExecution.createStepExecution(factory.getName());
+		repository.add(stepExecution);
+
+		Step step = (Step) factory.getObject();
+
+		step.execute(stepExecution);
+		assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
+
+		assertEquals("[]", writer.getCommitted().toString());
+	}
+
 	@SuppressWarnings("unchecked")
 	private Collection<Class<? extends Throwable>> getExceptionList(Class<? extends Throwable> arg) {
 		return Arrays.<Class<? extends Throwable>> asList(arg);
@@ -480,5 +519,5 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
 		}
 		return map;
 	}
-	
+
 }