diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
index 100d83721..65ce0c459 100755
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
@@ -236,7 +236,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 							// allows us to continue
 							throw e;
 						}
-						else if (itemProcessSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
+						else if (shouldSkip(itemProcessSkipPolicy, e, contribution.getStepSkipCount())) {
 							// If we are not re-throwing then we should check if
 							// this is skippable
 							contribution.incrementProcessSkipCount();
@@ -268,7 +268,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 
 				public O recover(RetryContext context) throws Exception {
 					Throwable e = context.getLastThrowable();
-					if (itemProcessSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
+					if (shouldSkip(itemProcessSkipPolicy, e, contribution.getStepSkipCount())) {
 						contribution.incrementProcessSkipCount();
 						iterator.remove(e);
 						logger.debug("Skipping after failed process", e);
@@ -376,7 +376,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 					 * do any scanning. We can just bomb out with a retry
 					 * exhausted.
 					 */
-					if (!itemWriteSkipPolicy.shouldSkip(context.getLastThrowable(), -1)) {
+					if (!shouldSkip(itemWriteSkipPolicy, context.getLastThrowable(), -1)) {
 						throw new ExhaustedRetryException(
 								"Retry exhausted after last attempt in recovery path, but exception is not skippable.",
 								context.getLastThrowable());
@@ -442,6 +442,23 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 		}
 	}
 
+	/**
+	 * Convenience method for calling process skip policy, so that it can be
+	 * called from multiple places.
+	 * 
+	 * @param policy the skip policy
+	 * @param e the cause of the skip
+	 * @param skipCount the current skip count
+	 */
+	private boolean shouldSkip(SkipPolicy policy, Throwable e, int skipCount) {
+		try {
+			return policy.shouldSkip(e, skipCount);
+		}
+		catch (RuntimeException ex) {
+			throw new SkipListenerFailedException("Fatal exception in SkipPolicy.", ex, e);
+		}
+	}
+
 	private Object getInputKey(I item) {
 		if (keyGenerator == null) {
 			return item;
@@ -463,7 +480,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 	private void checkSkipPolicy(Chunk<I>.ChunkIterator inputIterator, Chunk<O>.ChunkIterator outputIterator,
 			Throwable e, StepContribution contribution) {
 		logger.debug("Checking skip policy after failed write");
-		if (itemWriteSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
+		if (shouldSkip(itemWriteSkipPolicy, e, contribution.getStepSkipCount())) {
 			contribution.incrementWriteSkipCount();
 			inputIterator.remove();
 			outputIterator.remove(e);
@@ -498,7 +515,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 			outputIterator.remove();
 		}
 		catch (Exception e) {
-			if (!itemWriteSkipPolicy.shouldSkip(e, -1) && !rollbackClassifier.classify(e)) {
+			if (!shouldSkip(itemWriteSkipPolicy, e, -1) && !rollbackClassifier.classify(e)) {
 				inputIterator.remove();
 				outputIterator.remove();
 			}
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProvider.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProvider.java
index 8ccb38d91..2425d2611 100755
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProvider.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProvider.java
@@ -69,7 +69,7 @@ public class FaultTolerantChunkProvider<I> extends SimpleChunkProvider<I> {
 			}
 			catch (Exception e) {
 
-				if (skipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
+				if (shouldSkip(skipPolicy, e, contribution.getStepSkipCount())) {
 					// increment skip count and try again
 					contribution.incrementReadSkipCount();
 					chunk.skip(e);
@@ -99,4 +99,20 @@ public class FaultTolerantChunkProvider<I> extends SimpleChunkProvider<I> {
 		}
 	}
 
+	/**
+	 * Convenience method for calling process skip policy.
+	 * 
+	 * @param policy the skip policy
+	 * @param e the cause of the skip
+	 * @param skipCount the current skip count
+	 */
+	private boolean shouldSkip(SkipPolicy policy, Throwable e, int skipCount) {
+		try {
+			return policy.shouldSkip(e, skipCount);
+		}
+		catch (RuntimeException ex) {
+			throw new SkipListenerFailedException("Fatal exception in SkipPolicy.", ex, e);
+		}
+	}
+
 }
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 50e8b725b..c263f8587 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
@@ -35,6 +35,7 @@ import org.springframework.batch.core.step.skip.NonSkippableReadException;
 import org.springframework.batch.core.step.skip.SkipLimitExceededException;
 import org.springframework.batch.core.step.skip.SkipListenerFailedException;
 import org.springframework.batch.core.step.skip.SkipPolicy;
+import org.springframework.batch.core.step.skip.SkipPolicyFailedException;
 import org.springframework.batch.core.step.tasklet.TaskletStep;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemStream;
@@ -321,9 +322,9 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 	@Override
 	protected void applyConfiguration(TaskletStep step) {
 		addNonSkippableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
-				SkipListenerFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
+				SkipListenerFailedException.class, SkipPolicyFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
 		addNonRetryableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
-				SkipListenerFailedException.class, RetryException.class, JobInterruptedException.class, Error.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/skip/SkipPolicyFailedException.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/SkipPolicyFailedException.java
new file mode 100644
index 000000000..25ccc75cc
--- /dev/null
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/SkipPolicyFailedException.java
@@ -0,0 +1,39 @@
+/*
+ * Copyright 2006-2007 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.core.UnexpectedJobExecutionException;
+
+/**
+ * Special exception to indicate a failure in a skip policy. These need
+ * special treatment in the framework in case a skip sends itself into an
+ * infinite loop.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class SkipPolicyFailedException extends UnexpectedJobExecutionException {
+
+	/**
+	 * @param message describes the error to the user
+	 * @param ex the exception that was thrown by a {@link SkipPolicy}
+	 * @param t the exception that caused the skip
+	 */
+	public SkipPolicyFailedException(String message, RuntimeException ex, Throwable t) {
+		super(message + "\n" + t.getClass().getName() + ": " + t.getMessage(), ex);
+	}
+
+}
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 2423186f9..84958f091 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
@@ -34,6 +34,7 @@ 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.SkipLimitExceededException;
 import org.springframework.batch.core.step.skip.SkipPolicy;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemReader;
@@ -202,6 +203,60 @@ public class FaultTolerantStepFactoryBeanTests {
 		testReadSkip();
 	}
 
+	/**
+	 * Check items causing errors are skipped as expected.
+	 */
+	@Test
+	public void testReadSkipWithPolicyExceptionInReader() throws Exception {
+
+		// Should be ignored
+		factory.setSkipLimit(0);
+
+		factory.setSkipPolicy(new SkipPolicy() {
+			public boolean shouldSkip(Throwable t, int skipCount) throws SkipLimitExceededException {
+				throw new  RuntimeException("Planned exception in SkipPolicy");
+			}
+		});
+
+		reader.setFailures("2");
+
+		Step step = (Step) factory.getObject();
+
+		step.execute(stepExecution);
+		
+		assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
+		assertEquals(0, stepExecution.getReadSkipCount());
+		assertEquals(1, stepExecution.getReadCount());
+
+	}
+
+	/**
+	 * Check items causing errors are skipped as expected.
+	 */
+	@Test
+	public void testReadSkipWithPolicyExceptionInWriter() throws Exception {
+
+		// Should be ignored
+		factory.setSkipLimit(0);
+
+		factory.setSkipPolicy(new SkipPolicy() {
+			public boolean shouldSkip(Throwable t, int skipCount) throws SkipLimitExceededException {
+				throw new  RuntimeException("Planned exception in SkipPolicy");
+			}
+		});
+
+		writer.setFailures("2");
+
+		Step step = (Step) factory.getObject();
+
+		step.execute(stepExecution);
+		
+		assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
+		assertEquals(0, stepExecution.getWriteSkipCount());
+		assertEquals(2, stepExecution.getReadCount());
+
+	}
+
 	/**
 	 * Check to make sure that ItemStreamException can be skipped. (see
 	 * BATCH-915)
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/skip/SkipPolicyFailedExceptionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/skip/SkipPolicyFailedExceptionTests.java
new file mode 100644
index 000000000..85dbf1f0f
--- /dev/null
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/skip/SkipPolicyFailedExceptionTests.java
@@ -0,0 +1,35 @@
+/*
+ * Copyright 2006-2007 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.core.listener.AbstractDoubleExceptionTests;
+
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class SkipPolicyFailedExceptionTests extends AbstractDoubleExceptionTests {
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.core.listener.AbstractDoubleExceptionTests#getException(java.lang.String, java.lang.RuntimeException, java.lang.Throwable)
+	 */
+	@Override
+	public Exception getException(String msg, RuntimeException cause, Throwable e) throws Exception {
+		return new SkipPolicyFailedException(msg, cause, e);
+	}
+
+}