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 670e2f1db..93501562d 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,9 @@ 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.ExceptionClassifierSkipPolicy;
 import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
+import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy;
 import org.springframework.batch.core.step.skip.NonSkippableReadException;
 import org.springframework.batch.core.step.skip.SkipLimitExceededException;
 import org.springframework.batch.core.step.skip.SkipListenerFailedException;
@@ -76,6 +78,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 
 	private Map<Class<? extends Throwable>, Boolean> skippableExceptionClasses = new HashMap<Class<? extends Throwable>, Boolean>();
 
+	private Collection<Class<? extends Throwable>> nonSkippableExceptionClasses = new HashSet<Class<? extends Throwable>>();
+
 	private Collection<Class<? extends Throwable>> noRollbackExceptionClasses = new HashSet<Class<? extends Throwable>>();
 
 	private Map<Class<? extends Throwable>, Boolean> retryableExceptionClasses = new HashMap<Class<? extends Throwable>, Boolean>();
@@ -88,6 +92,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 
 	private int skipLimit = 0;
 
+	private SkipPolicy skipPolicy;
+
 	private BackOffPolicy backOffPolicy;
 
 	private RetryListener[] retryListeners;
@@ -207,6 +213,18 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 		this.skipLimit = skipLimit;
 	}
 
+	/**
+	 * A {@link SkipPolicy} that determines the outcome of an exception when
+	 * processing an item. Overrides the {@link #setSkipLimit(int) skipLimit}.
+	 * The {@link #setSkippableExceptionClasses(Map) skippableExceptionClasses}
+	 * are also ignored if this is set.
+	 * 
+	 * @param skipPolicy the {@link SkipPolicy} to set
+	 */
+	public void setSkipPolicy(SkipPolicy skipPolicy) {
+		this.skipPolicy = skipPolicy;
+	}
+
 	/**
 	 * Exception classes that when raised won't crash the job but will result in
 	 * the item which handling caused the exception being skipped. Any exception
@@ -300,14 +318,12 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 
 	}
 
-	@SuppressWarnings("unchecked")
 	@Override
 	protected void applyConfiguration(TaskletStep step) {
-		addFatalExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
+		addNonSkippableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
 				SkipListenerFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
 		addNonRetryableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
 				SkipListenerFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
-
 		super.applyConfiguration(step);
 	}
 
@@ -351,7 +367,9 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 	@Override
 	protected SimpleChunkProvider<T> configureChunkProvider() {
 
-		SkipPolicy readSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, getSkippableExceptionClasses());
+		SkipPolicy readSkipPolicy = skipPolicy != null ? skipPolicy : new LimitCheckingItemSkipPolicy(skipLimit,
+				getSkippableExceptionClasses());
+		readSkipPolicy = getFatalExceptionAwareProxy(readSkipPolicy);
 		FaultTolerantChunkProvider<T> chunkProvider = new FaultTolerantChunkProvider<T>(getItemReader(),
 				getChunkOperations());
 		chunkProvider.setSkipPolicy(readSkipPolicy);
@@ -374,7 +392,9 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 		chunkProcessor.setBuffering(!isReaderTransactionalQueue());
 		chunkProcessor.setProcessorTransactional(processorTransactional);
 
-		SkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, getSkippableExceptionClasses());
+		SkipPolicy writeSkipPolicy = skipPolicy != null ? skipPolicy : new LimitCheckingItemSkipPolicy(skipLimit,
+				getSkippableExceptionClasses());
+		writeSkipPolicy = getFatalExceptionAwareProxy(writeSkipPolicy);
 		chunkProcessor.setWriteSkipPolicy(writeSkipPolicy);
 		chunkProcessor.setProcessSkipPolicy(writeSkipPolicy);
 		chunkProcessor.setRollbackClassifier(getRollbackClassifier());
@@ -409,7 +429,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 			retryPolicy = new SimpleRetryPolicy(retryLimit, map);
 		}
 
-		RetryPolicy retryPolicyWrapper = fatalExceptionAwareProxy(retryPolicy);
+		RetryPolicy retryPolicyWrapper = getFatalExceptionAwareProxy(retryPolicy);
 
 		BatchRetryTemplate batchRetryTemplate = new BatchRetryTemplate();
 		if (backOffPolicy != null) {
@@ -444,7 +464,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 	 * Wrap the provided {@link #setRetryPolicy(RetryPolicy)} so that it never
 	 * retries explicitly non-retryable exceptions.
 	 */
-	private RetryPolicy fatalExceptionAwareProxy(final RetryPolicy retryPolicy) {
+	private RetryPolicy getFatalExceptionAwareProxy(RetryPolicy retryPolicy) {
 
 		NeverRetryPolicy neverRetryPolicy = new NeverRetryPolicy();
 		Map<Class<? extends Throwable>, RetryPolicy> map = new HashMap<Class<? extends Throwable>, RetryPolicy>();
@@ -462,15 +482,41 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 
 	}
 
-	private void addFatalExceptionIfMissing(Class<? extends Throwable>... classes) {
-		Map<Class<? extends Throwable>, Boolean> exceptions = new HashMap<Class<? extends Throwable>, Boolean>(
-				skippableExceptionClasses);
-		for (Class<? extends Throwable> cls : classes) {
-			if (!exceptions.containsKey(cls)) {
-				exceptions.put(cls, false);
+	/**
+	 * Wrap a {@link SkipPolicy} and make it consistent with known fatal
+	 * exceptions.
+	 * 
+	 * @param skipPolicy an existing skip policy
+	 * @return a skip policy that will not skip fatal exceptions
+	 */
+	private SkipPolicy getFatalExceptionAwareProxy(SkipPolicy skipPolicy) {
+
+		NeverSkipItemSkipPolicy neverSkipPolicy = new NeverSkipItemSkipPolicy();
+		Map<Class<? extends Throwable>, SkipPolicy> map = new HashMap<Class<? extends Throwable>, SkipPolicy>();
+		for (Class<? extends Throwable> fatal : nonSkippableExceptionClasses) {
+			map.put(fatal, neverSkipPolicy);
+		}
+
+		SubclassClassifier<Throwable, SkipPolicy> classifier = new SubclassClassifier<Throwable, SkipPolicy>(skipPolicy);
+		classifier.setTypeMap(map);
+
+		ExceptionClassifierSkipPolicy skipPolicyWrapper = new ExceptionClassifierSkipPolicy();
+		skipPolicyWrapper.setExceptionClassifier(classifier);
+		return skipPolicyWrapper;
+	}
+
+	@SuppressWarnings("unchecked")
+	private void addNonSkippableExceptionIfMissing(Class... cls) {
+		List exceptions = new ArrayList<Class<? extends Throwable>>();
+		for (Class exceptionClass : nonSkippableExceptionClasses) {
+			exceptions.add(exceptionClass);
+		}
+		for (Class fatal : cls) {
+			if (!exceptions.contains(fatal)) {
+				exceptions.add(fatal);
 			}
 		}
-		skippableExceptionClasses = exceptions;
+		nonSkippableExceptionClasses = exceptions;
 	}
 
 	@SuppressWarnings("unchecked")
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/ExceptionClassifierSkipPolicy.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/ExceptionClassifierSkipPolicy.java
new file mode 100644
index 000000000..6108927eb
--- /dev/null
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/ExceptionClassifierSkipPolicy.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright 2006-2010 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.step.skip;
+
+import org.springframework.batch.classify.SubclassClassifier;
+
+/**
+ * A {@link SkipPolicy} that depends on an exception classifier to make its
+ * decision, and then delegates to the classifier result.
+ * 
+ * @author Dave Syer
+ * 
+ * @see SubclassClassifier
+ */
+public class ExceptionClassifierSkipPolicy implements SkipPolicy {
+
+	private SubclassClassifier<Throwable, SkipPolicy> classifier;
+
+	/**
+	 * The classifier that will be used to choose a delegate policy.
+	 * 
+	 * @param classifier the classifier to use to choose a delegate policy
+	 */
+	public void setExceptionClassifier(SubclassClassifier<Throwable, SkipPolicy> classifier) {
+		this.classifier = classifier;
+	}
+
+	/**
+	 * Consult the classifier and find a delegate policy, and then use that to
+	 * determine the outcome.
+	 * 
+	 * @param t the throwable to consider
+	 * @param skipCount the current skip count
+	 * @return true if the exception can be skipped
+	 * @throws SkipLimitExceededException if a limit is exceeded
+	 */
+	public boolean shouldSkip(Throwable t, int skipCount) throws SkipLimitExceededException {
+		return classifier.classify(t).shouldSkip(t, skipCount);
+	}
+
+}
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 c8279cbf9..9f3746001 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
@@ -26,8 +26,10 @@ import java.util.Collection;
 import java.util.Map;
 
 import org.junit.Test;
+import org.springframework.batch.classify.SubclassClassifier;
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.step.item.SimpleChunkProcessor;
+import org.springframework.batch.core.step.skip.SkipPolicy;
 import org.springframework.batch.core.step.tasklet.TaskletStep;
 import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.support.CompositeItemStream;
@@ -103,7 +105,7 @@ public class ChunkElementParserTests {
 	@Test
 	public void testInheritSkippable() throws Exception {
 		Map<Class<? extends Throwable>, Boolean> skippable = getExceptionClasses("s1", getContext());
-		assertEquals(11, skippable.size());
+		assertEquals(5, skippable.size());
 		containsClassified(skippable, NullPointerException.class, true);
 		containsClassified(skippable, ArithmeticException.class, true);
 		containsClassified(skippable, CannotAcquireLockException.class, false);
@@ -113,7 +115,7 @@ public class ChunkElementParserTests {
 	@Test
 	public void testInheritSkippableWithNoMerge() throws Exception {
 		Map<Class<? extends Throwable>, Boolean> skippable = getExceptionClasses("s2", getContext());
-		assertEquals(9, skippable.size());
+		assertEquals(3, skippable.size());
 		containsClassified(skippable, NullPointerException.class, true);
 		assertFalse(skippable.containsKey(ArithmeticException.class));
 		containsClassified(skippable, CannotAcquireLockException.class, false);
@@ -194,8 +196,11 @@ public class ChunkElementParserTests {
 		Object tasklet = ReflectionTestUtils.getField(step, "tasklet");
 		Object chunkProvider = ReflectionTestUtils.getField(tasklet, "chunkProvider");
 		Object skipPolicy = ReflectionTestUtils.getField(chunkProvider, "skipPolicy");
-		Object classifier = ReflectionTestUtils.getField(skipPolicy, "skippableExceptionClassifier");
-		return (Map<Class<? extends Throwable>, Boolean>) ReflectionTestUtils.getField(classifier, "classified");
+		SubclassClassifier<Throwable, SkipPolicy> classifier = (SubclassClassifier<Throwable, SkipPolicy>) ReflectionTestUtils
+				.getField(skipPolicy, "classifier");
+		Object limitPolicy = classifier.classify(new Exception());
+		Object limitClassifier = ReflectionTestUtils.getField(limitPolicy, "skippableExceptionClassifier");
+		return (Map<Class<? extends Throwable>, Boolean>) ReflectionTestUtils.getField(limitClassifier, "classified");
 	}
 
 	@SuppressWarnings("unchecked")
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 e0d295c4f..81d12f1f9 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
@@ -6,6 +6,7 @@ import static org.junit.Assert.assertTrue;
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -32,6 +33,7 @@ import org.springframework.batch.core.StepListener;
 import org.springframework.batch.core.listener.SkipListenerSupport;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
+import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
 import org.springframework.batch.core.step.skip.SkipPolicy;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemReader;
@@ -188,6 +190,18 @@ public class FaultTolerantStepFactoryBeanTests {
 				.getName()));
 	}
 
+	/**
+	 * Check items causing errors are skipped as expected.
+	 */
+	@Test
+	public void testReadSkipWithPolicy() throws Exception {
+		// Should be ignored
+		factory.setSkipLimit(0);
+		factory.setSkipPolicy(new LimitCheckingItemSkipPolicy(2, Collections
+				.<Class<? extends Throwable>, Boolean> singletonMap(Exception.class, true)));
+		testReadSkip();
+	}
+
 	/**
 	 * Check to make sure that ItemStreamException can be skipped. (see
 	 * BATCH-915)
@@ -426,7 +440,7 @@ public class FaultTolerantStepFactoryBeanTests {
 		step.execute(stepExecution);
 
 		// 1,3 skipped inside a committed chunk. 5 tripped the skip
-		// limit but it was skipped in a chunk that rolled back, so 
+		// limit but it was skipped in a chunk that rolled back, so
 		// it will re-appear on a restart and the listener is not called.
 		assertEquals(2, listenerCalls.size());
 		assertEquals(2, stepExecution.getReadSkipCount());
diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserParentAttributeTests-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserParentAttributeTests-context.xml
index 1489c9133..ce460680d 100644
--- a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserParentAttributeTests-context.xml
+++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserParentAttributeTests-context.xml
@@ -45,7 +45,7 @@
 	
 	<step id="standalone2" parent="baseStep">
 		<tasklet>
-			<chunk reader="reader" writer="writer" commit-interval="5" />
+			<chunk reader="reader" writer="writer" commit-interval="5"/>
 			<transaction-attributes propagation="REQUIRED" isolation="DEFAULT" />
 		</tasklet>
 	</step>