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 10bfa5f8d..99035b361 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
@@ -18,11 +18,14 @@ package org.springframework.batch.core.step.item;
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 
 import org.springframework.batch.classify.BinaryExceptionClassifier;
 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.skip.LimitCheckingItemSkipPolicy;
@@ -77,6 +80,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 
 	private Collection<Class<? extends Throwable>> fatalExceptionClasses = new HashSet<Class<? extends Throwable>>();
 
+	private Collection<Class<? extends Throwable>> nonRetryableExceptionClasses = new HashSet<Class<? extends Throwable>>();
+
 	private Collection<Class<? extends Throwable>> retryableExceptionClasses = new HashSet<Class<? extends Throwable>>();
 
 	private int cacheCapacity = 0;
@@ -228,7 +233,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 	}
 
 	/**
-	 * Exception classes that should cause immediate failure.
+	 * Exception classes that are not skippable (but may be retryable).
 	 * 
 	 * @param fatalExceptionClasses {@link Error} by default
 	 */
@@ -296,6 +301,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 
 		addFatalExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
 				SkipListenerFailedException.class, RetryException.class, JobInterruptedException.class);
+		addNonRetryableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
+				SkipListenerFailedException.class, RetryException.class, JobInterruptedException.class);
 
 		super.applyConfiguration(step);
 
@@ -358,14 +365,14 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 	@Override
 	protected SimpleChunkProcessor<T, S> configureChunkProcessor() {
 
-
 		BatchRetryTemplate batchRetryTemplate = configureRetry();
 
 		FaultTolerantChunkProcessor<T, S> chunkProcessor = new FaultTolerantChunkProcessor<T, S>(getItemProcessor(),
 				getItemWriter(), batchRetryTemplate);
 		chunkProcessor.setBuffering(!isReaderTransactionalQueue());
 
-		SkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, getSkippableExceptionClasses(), fatalExceptionClasses);
+		SkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, getSkippableExceptionClasses(),
+				fatalExceptionClasses);
 		chunkProcessor.setWriteSkipPolicy(writeSkipPolicy);
 		chunkProcessor.setProcessSkipPolicy(writeSkipPolicy);
 		chunkProcessor.setRollbackClassifier(getRollbackClassifier());
@@ -412,7 +419,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 		RepeatOperations stepOperations = getStepOperations();
 		if (stepOperations instanceof RepeatTemplate) {
 			SimpleRetryExceptionHandler exceptionHandler = new SimpleRetryExceptionHandler(retryPolicyWrapper,
-					getExceptionHandler(), fatalExceptionClasses);
+					getExceptionHandler(), nonRetryableExceptionClasses);
 			((RepeatTemplate) stepOperations).setExceptionHandler(exceptionHandler);
 		}
 
@@ -433,39 +440,52 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 
 	/**
 	 * Wrap the provided {@link #setRetryPolicy(RetryPolicy)} so that it never
-	 * retries fatal exceptions.
+	 * retries explicitly non-retryable exceptions.
 	 */
 	private RetryPolicy fatalExceptionAwareProxy(final RetryPolicy retryPolicy) {
 
-		final NeverRetryPolicy neverRetryPolicy = new NeverRetryPolicy();
+		NeverRetryPolicy neverRetryPolicy = new NeverRetryPolicy();
+		Map<Class<? extends Throwable>, RetryPolicy> map = new HashMap<Class<? extends Throwable>, RetryPolicy>();
+		for (Class<? extends Throwable> fatal : nonRetryableExceptionClasses) {
+			map.put(fatal, neverRetryPolicy);
+		}
+
+		SubclassClassifier<Throwable, RetryPolicy> classifier = new SubclassClassifier<Throwable, RetryPolicy>(
+				retryPolicy);
+		classifier.setTypeMap(map);
+
 		ExceptionClassifierRetryPolicy retryPolicyWrapper = new ExceptionClassifierRetryPolicy();
-		retryPolicyWrapper.setExceptionClassifier(new Classifier<Throwable, RetryPolicy>() {
-
-			public RetryPolicy classify(Throwable classifiable) {
-
-				for (Class<? extends Throwable> fatal : fatalExceptionClasses) {
-					if (fatal.isAssignableFrom(classifiable.getClass())) {
-						return neverRetryPolicy;
-					}
-				}
-				return retryPolicy;
-			}
-		});
+		retryPolicyWrapper.setExceptionClassifier(classifier);
 		return retryPolicyWrapper;
+
 	}
 
 	@SuppressWarnings("unchecked")
 	private void addFatalExceptionIfMissing(Class... cls) {
-		List fatalExceptionList = new ArrayList<Class<? extends Throwable>>();
+		List exceptions = new ArrayList<Class<? extends Throwable>>();
 		for (Class exceptionClass : fatalExceptionClasses) {
-			fatalExceptionList.add(exceptionClass);
+			exceptions.add(exceptionClass);
 		}
 		for (Class fatal : cls) {
-			if (!fatalExceptionList.contains(fatal)) {
-				fatalExceptionList.add(fatal);
+			if (!exceptions.contains(fatal)) {
+				exceptions.add(fatal);
 			}
 		}
-		fatalExceptionClasses = fatalExceptionList;
+		fatalExceptionClasses = exceptions;
+	}
+
+	@SuppressWarnings("unchecked")
+	private void addNonRetryableExceptionIfMissing(Class... cls) {
+		List exceptions = new ArrayList<Class<? extends Throwable>>();
+		for (Class exceptionClass : nonRetryableExceptionClasses) {
+			exceptions.add(exceptionClass);
+		}
+		for (Class fatal : cls) {
+			if (!exceptions.contains(fatal)) {
+				exceptions.add(fatal);
+			}
+		}
+		nonRetryableExceptionClasses = exceptions;
 	}
 
 }
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 a4a698dab..71c2ca496 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
@@ -164,13 +164,13 @@ public class FaultTolerantExceptionClassesTests implements ApplicationContextAwa
 	@Test
 	public void testRetryableFatal() throws Exception {
 		// User wants all exceptions to be retried, but only some are skippable
-		// FatalRuntimeException is not skippable, but is a subclass of another
-		// skippable
+		// FatalRuntimeException is not skippable because it is fatal, but is a
+		// subclass of another skippable
 		writer.setExceptionType(FatalRuntimeException.class);
 		StepExecution stepExecution = launchStep("retryable");
 		assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
-		// TODO BATCH-1333: assertEquals("[1, 2, 3, 1, 2, 3, 1, 2, 3]",
-		// writer.getWritten().toString());
+		// BATCH-1333:
+		assertEquals("[1, 2, 3, 1, 2, 3]", writer.getWritten().toString());
 		assertEquals("[]", writer.getCommitted().toString());
 	}
 
@@ -198,9 +198,10 @@ public class FaultTolerantExceptionClassesTests implements ApplicationContextAwa
 		writer.setExceptionType(FatalException.class);
 		StepExecution stepExecution = launchStep("retryable");
 		assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
-		// TODO BATCH-1333: assertEquals("[1, 2, 3, 1, 2, 3, 1, 2, 3]",
-		// writer.getWritten().toString());
+		// BATCH-1333:
+		assertEquals("[1, 2, 3, 1, 2, 3]", writer.getWritten().toString());
 		assertEquals("[]", writer.getCommitted().toString());
+		assertEquals(0, stepExecution.getWriteSkipCount());
 	}
 
 	@Test
@@ -259,11 +260,11 @@ public class FaultTolerantExceptionClassesTests implements ApplicationContextAwa
 		writer.setExceptionType(FatalRuntimeException.class);
 		StepExecution stepExecution = launchStep("noRollbackSkippable");
 		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
+		// BATCH-1332:
 		assertEquals("[1, 2, 3, 1, 2, 3, 4]", writer.getWritten().toString());
-		// BATCH-1332: 
+		// TODO BATCH-1334:
+		// Skipped but also committed (because it was marked as no-rollback)
 		assertEquals("[1, 2, 3, 4]", writer.getCommitted().toString());
-		// TODO BATCH-1334: 
-		// Not skipped but also committed (because it was marked as no-rollback)
 		assertEquals(1, stepExecution.getWriteSkipCount());
 	}
 
@@ -279,16 +280,17 @@ public class FaultTolerantExceptionClassesTests implements ApplicationContextAwa
 	@Test
 	public void testNoRollbackFatalNoRollbackException() throws Exception {
 		// User has asked for no rollback on a fatal exception. What should the
-		// outcome be?
+		// outcome be?  As per BATCH-1333 it is interpreted as not skippable, but
+		// retryable if requested.  Here it was not requested to be retried, but
+		// it was marked as no-rollback.  As per BATCH-1334 this has to be ignored
+		// so that the failed item can be isolated.
 		writer.setExceptionType(FatalRuntimeException.class);
 		StepExecution stepExecution = launchStep("noRollbackFatal");
 		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
-		// TODO BATCH-1331: assertEquals(BatchStatus.FAILED,
-		// stepExecution.getStatus());
-		// TODO BATCH-1331: assertEquals("[1, 2, 3]",
-		// writer.getWritten().toString());
-		// TODO BATCH-1331: assertEquals("[1, 2, 3]",
-		// writer.getCommitted().toString());
+		// BATCH-1331: 
+		assertEquals("[1, 2, 3, 1, 2, 3, 4]", writer.getWritten().toString());
+		// BATCH-1331: 
+		assertEquals("[1, 2, 3, 4]", writer.getCommitted().toString());
 	}
 
 	@Test
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java
index a1eb35821..861aa4f7b 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java
@@ -279,8 +279,9 @@ public class FaultTolerantStepFactoryBeanTests {
 	}
 
 	/**
-	 * Fatal exception should cause immediate termination regardless of other
-	 * skip settings (note the fatal exception is also classified as rollback).
+	 * Fatal exception should cause immediate termination provided the exception
+	 * is not skippable (note the fatal exception is also classified as
+	 * rollback).
 	 */
 	@Test
 	public void testFatalException() throws Exception {
@@ -296,7 +297,8 @@ public class FaultTolerantStepFactoryBeanTests {
 		Step step = (Step) factory.getObject();
 
 		step.execute(stepExecution);
-		assertTrue(stepExecution.getFailureExceptions().get(0).getMessage().equals("Ouch!"));
+		String message = stepExecution.getFailureExceptions().get(0).getCause().getMessage();
+		assertTrue("Wrong message: " + message, message.equals("Ouch!"));
 		assertStepExecutionsAreEqual(stepExecution, repository.getLastStepExecution(jobExecution.getJobInstance(), step
 				.getName()));
 	}