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 0abf493c6..e03a270cf 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
@@ -46,7 +46,8 @@ import org.springframework.batch.retry.support.DefaultRetryState;
  * allows for skipping or retry of items that cause exceptions during writing.
  * 
  */
-public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O> {
+public class FaultTolerantChunkProcessor<I, O> extends
+		SimpleChunkProcessor<I, O> {
 
 	private SkipPolicy itemProcessSkipPolicy = new LimitCheckingItemSkipPolicy();
 
@@ -54,7 +55,8 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 
 	private final BatchRetryTemplate batchRetryTemplate;
 
-	private Classifier<Throwable, Boolean> rollbackClassifier = new BinaryExceptionClassifier(true);
+	private Classifier<Throwable, Boolean> rollbackClassifier = new BinaryExceptionClassifier(
+			true);
 
 	private Log logger = LogFactory.getLog(getClass());
 
@@ -71,21 +73,24 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 	 * Not used in the case of the {@link #setBuffering(boolean) buffering flag}
 	 * being true (the default).
 	 * 
-	 * @param keyGenerator the {@link KeyGenerator} to set
+	 * @param keyGenerator
+	 *            the {@link KeyGenerator} to set
 	 */
 	public void setKeyGenerator(KeyGenerator keyGenerator) {
 		this.keyGenerator = keyGenerator;
 	}
 
 	/**
-	 * @param SkipPolicy the {@link SkipPolicy} for item processing
+	 * @param SkipPolicy
+	 *            the {@link SkipPolicy} for item processing
 	 */
 	public void setProcessSkipPolicy(SkipPolicy SkipPolicy) {
 		this.itemProcessSkipPolicy = SkipPolicy;
 	}
 
 	/**
-	 * @param SkipPolicy the {@link SkipPolicy} for item writing
+	 * @param SkipPolicy
+	 *            the {@link SkipPolicy} for item writing
 	 */
 	public void setWriteSkipPolicy(SkipPolicy SkipPolicy) {
 		this.itemWriteSkipPolicy = SkipPolicy;
@@ -97,7 +102,8 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 	 * 
 	 * @param rollbackClassifier
 	 */
-	public void setRollbackClassifier(Classifier<Throwable, Boolean> rollbackClassifier) {
+	public void setRollbackClassifier(
+			Classifier<Throwable, Boolean> rollbackClassifier) {
 		this.rollbackClassifier = rollbackClassifier;
 	}
 
@@ -125,14 +131,17 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 	 * true). If false then the processor is only called once per item per
 	 * chunk, even if there are rollbacks with retries and skips.
 	 * 
-	 * @param processorTransactional the flag value to set
+	 * @param processorTransactional
+	 *            the flag value to set
 	 */
 	public void setProcessorTransactional(boolean processorTransactional) {
 		this.processorTransactional = processorTransactional;
 	}
 
-	public FaultTolerantChunkProcessor(ItemProcessor<? super I, ? extends O> itemProcessor,
-			ItemWriter<? super O> itemWriter, BatchRetryTemplate batchRetryTemplate) {
+	public FaultTolerantChunkProcessor(
+			ItemProcessor<? super I, ? extends O> itemProcessor,
+			ItemWriter<? super O> itemWriter,
+			BatchRetryTemplate batchRetryTemplate) {
 		super(itemProcessor, itemWriter);
 		this.batchRetryTemplate = batchRetryTemplate;
 	}
@@ -193,16 +202,19 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 	}
 
 	@Override
-	protected Chunk<O> transform(final StepContribution contribution, Chunk<I> inputs) throws Exception {
+	protected Chunk<O> transform(final StepContribution contribution,
+			Chunk<I> inputs) throws Exception {
 
 		Chunk<O> outputs = new Chunk<O>();
 		@SuppressWarnings("unchecked")
 		final UserData<O> data = (UserData<O>) inputs.getUserData();
 		final Chunk<O> cache = data.getOutputs();
-		final Iterator<O> cacheIterator = cache.isEmpty() ? null : new ArrayList<O>(cache.getItems()).iterator();
+		final Iterator<O> cacheIterator = cache.isEmpty() ? null
+				: new ArrayList<O>(cache.getItems()).iterator();
 		final AtomicInteger count = new AtomicInteger(0);
 
-		for (final Chunk<I>.ChunkIterator iterator = inputs.iterator(); iterator.hasNext();) {
+		for (final Chunk<I>.ChunkIterator iterator = inputs.iterator(); iterator
+				.hasNext();) {
 
 			final int scanLimit = processorTransactional ? 1 : 0;
 			final I item = iterator.next();
@@ -213,7 +225,8 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 					O output = null;
 					try {
 						count.incrementAndGet();
-						O cached = (cacheIterator != null && cacheIterator.hasNext()) ? cacheIterator.next() : null;
+						O cached = (cacheIterator != null && cacheIterator
+								.hasNext()) ? cacheIterator.next() : null;
 						if (cached != null && count.get() > scanLimit) {
 							/*
 							 * If there is a cached chunk then we must be
@@ -223,30 +236,29 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 							 * cache.
 							 */
 							output = cached;
-						}
-						else {
+						} else {
 							output = doProcess(item);
 							if (!processorTransactional) {
 								cache.add(output);
 							}
 						}
-					}
-					catch (Exception e) {
+					} catch (Exception e) {
 						if (rollbackClassifier.classify(e)) {
 							// Default is to rollback unless the classifier
 							// allows us to continue
 							throw e;
-						}
-						else if (shouldSkip(itemProcessSkipPolicy, 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();
-							logger.debug("Skipping after failed process with no rollback", e);
+							logger.debug(
+									"Skipping after failed process with no rollback",
+									e);
 							// If not re-throwing then the listener will not be
 							// called in next chunk.
 							callProcessSkipListener(item, e);
-						}
-						else {
+						} else {
 							// If it's not skippable that's an error in
 							// configuration - it doesn't make sense to not roll
 							// back if we are also not allowed to skip
@@ -269,17 +281,19 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 
 				public O recover(RetryContext context) throws Exception {
 					Throwable e = context.getLastThrowable();
-					if (shouldSkip(itemProcessSkipPolicy, e, contribution.getStepSkipCount())) {
+					if (shouldSkip(itemProcessSkipPolicy, e,
+							contribution.getStepSkipCount())) {
 						iterator.remove(e);
 						contribution.incrementProcessSkipCount();
 						logger.debug("Skipping after failed process", e);
 						return null;
-					}
-					else {
+					} else {
 						if (rollbackClassifier.classify(e)) {
 							// Default is to rollback unless the classifier
 							// allows us to continue
-							throw new RetryException("Non-skippable exception in recoverer while processing", e);
+							throw new RetryException(
+									"Non-skippable exception in recoverer while processing",
+									e);
 						}
 						iterator.remove(e);
 						return null;
@@ -288,8 +302,9 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 
 			};
 
-			O output = batchRetryTemplate.execute(retryCallback, recoveryCallback, new DefaultRetryState(
-					getInputKey(item), rollbackClassifier));
+			O output = batchRetryTemplate.execute(retryCallback,
+					recoveryCallback, new DefaultRetryState(getInputKey(item),
+							rollbackClassifier));
 			if (output != null) {
 				outputs.add(output);
 			}
@@ -301,8 +316,8 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 	}
 
 	@Override
-	protected void write(final StepContribution contribution, final Chunk<I> inputs, final Chunk<O> outputs)
-			throws Exception {
+	protected void write(final StepContribution contribution,
+			final Chunk<I> inputs, final Chunk<O> outputs) throws Exception {
 
 		RetryCallback<Object> retryCallback = new RetryCallback<Object>() {
 			public Object doWithRetry(RetryContext context) throws Exception {
@@ -311,8 +326,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 					chunkMonitor.setChunkSize(inputs.size());
 					try {
 						doWrite(outputs.getItems());
-					}
-					catch (Exception e) {
+					} catch (Exception e) {
 						if (rollbackClassifier.classify(e)) {
 							throw e;
 						}
@@ -323,12 +337,12 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 						 * contract.
 						 */
 						throw new ForceRollbackForWriteSkipException(
-								"Force rollback on skippable exception so that skipped item can be located.", e);
+								"Force rollback on skippable exception so that skipped item can be located.",
+								e);
 					}
 					contribution.incrementWriteCount(outputs.size());
-				}
-				else {
-					scan(contribution, inputs, outputs, chunkMonitor);
+				} else {
+					scan(contribution, inputs, outputs, chunkMonitor, false);
 				}
 				return null;
 
@@ -343,17 +357,21 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 
 					Throwable e = context.getLastThrowable();
 					if (outputs.size() > 1 && !rollbackClassifier.classify(e)) {
-						throw new RetryException("Invalid retry state during write caused by "
-								+ "exception that does not classify for rollback: ", e);
+						throw new RetryException(
+								"Invalid retry state during write caused by "
+										+ "exception that does not classify for rollback: ",
+								e);
 					}
 
 					Chunk<I>.ChunkIterator inputIterator = inputs.iterator();
-					for (Chunk<O>.ChunkIterator outputIterator = outputs.iterator(); outputIterator.hasNext();) {
+					for (Chunk<O>.ChunkIterator outputIterator = outputs
+							.iterator(); outputIterator.hasNext();) {
 
 						inputIterator.next();
 						outputIterator.next();
 
-						checkSkipPolicy(inputIterator, outputIterator, e, contribution);
+						checkSkipPolicy(inputIterator, outputIterator, e,
+								contribution, true);
 						if (!rollbackClassifier.classify(e)) {
 							throw new RetryException(
 									"Invalid retry state during recovery caused by exception that does not classify for rollback: ",
@@ -369,10 +387,10 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 			};
 
 			batchRetryTemplate.execute(retryCallback, batchRecoveryCallback,
-					BatchRetryTemplate.createState(getInputKeys(inputs), rollbackClassifier));
+					BatchRetryTemplate.createState(getInputKeys(inputs),
+							rollbackClassifier));
 
-		}
-		else {
+		} else {
 
 			RecoveryCallback<Object> recoveryCallback = new RecoveryCallback<Object>() {
 
@@ -383,14 +401,15 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 					 * do any scanning. We can just bomb out with a retry
 					 * exhausted.
 					 */
-					if (!shouldSkip(itemWriteSkipPolicy, 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());
 					}
 
 					inputs.setBusy(true);
-					scan(contribution, inputs, outputs, chunkMonitor);
+					scan(contribution, inputs, outputs, chunkMonitor, true);
 					return null;
 				}
 
@@ -399,8 +418,8 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 			if (logger.isDebugEnabled()) {
 				logger.debug("Attempting to write: " + inputs);
 			}
-			batchRetryTemplate.execute(retryCallback, recoveryCallback, new DefaultRetryState(inputs,
-					rollbackClassifier));
+			batchRetryTemplate.execute(retryCallback, recoveryCallback,
+					new DefaultRetryState(inputs, rollbackClassifier));
 
 		}
 
@@ -423,9 +442,9 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 			Throwable e = wrapper.getException();
 			try {
 				getListener().onSkipInWrite(wrapper.getItem(), e);
-			}
-			catch (RuntimeException ex) {
-				throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
+			} catch (RuntimeException ex) {
+				throw new SkipListenerFailedException(
+						"Fatal exception in SkipListener.", ex, e);
 			}
 		}
 
@@ -439,15 +458,17 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 	 * Convenience method for calling process skip listener, so that it can be
 	 * called from multiple places.
 	 * 
-	 * @param item the item that is skipped
-	 * @param e the cause of the skip
+	 * @param item
+	 *            the item that is skipped
+	 * @param e
+	 *            the cause of the skip
 	 */
 	private void callProcessSkipListener(I item, Throwable e) {
 		try {
 			getListener().onSkipInProcess(item, e);
-		}
-		catch (RuntimeException ex) {
-			throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
+		} catch (RuntimeException ex) {
+			throw new SkipListenerFailedException(
+					"Fatal exception in SkipListener.", ex, e);
 		}
 	}
 
@@ -455,19 +476,21 @@ 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
+	 * @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 (SkipLimitExceededException ex) {
+		} catch (SkipLimitExceededException ex) {
 			throw ex;
-		}
-		catch (RuntimeException ex) {
-			throw new SkipListenerFailedException("Fatal exception in SkipPolicy.", ex, e);
+		} catch (RuntimeException ex) {
+			throw new SkipListenerFailedException(
+					"Fatal exception in SkipPolicy.", ex, e);
 		}
 	}
 
@@ -489,25 +512,44 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 		return keys;
 	}
 
-	private void checkSkipPolicy(Chunk<I>.ChunkIterator inputIterator, Chunk<O>.ChunkIterator outputIterator,
-			Throwable e, StepContribution contribution) {
+	private void checkSkipPolicy(Chunk<I>.ChunkIterator inputIterator,
+			Chunk<O>.ChunkIterator outputIterator, Throwable e,
+			StepContribution contribution, boolean recovery) throws Exception {
 		logger.debug("Checking skip policy after failed write");
 		if (shouldSkip(itemWriteSkipPolicy, e, contribution.getStepSkipCount())) {
 			contribution.incrementWriteSkipCount();
 			inputIterator.remove();
 			outputIterator.remove(e);
 			logger.debug("Skipping after failed write", e);
-		}
-		else {
-			throw new RetryException("Non-skippable exception in recoverer", e);
+		} else {
+			if (recovery) {
+				// Only if already recovering should we check skip policy
+				throw new RetryException(
+						"Non-skippable exception in recoverer", e);
+			} else {
+				if (e instanceof Exception) {
+					throw (Exception) e;
+				} else if (e instanceof Error) {
+					throw (Error) e;
+				} else {
+					throw new RetryException(
+							"Non-skippable throwable in recoverer", e);
+				}
+			}
 		}
 	}
 
-	private void scan(final StepContribution contribution, final Chunk<I> inputs, final Chunk<O> outputs,
-			ChunkMonitor chunkMonitor) throws Exception {
+	private void scan(final StepContribution contribution,
+			final Chunk<I> inputs, final Chunk<O> outputs,
+			ChunkMonitor chunkMonitor, boolean recovery) throws Exception {
 
 		if (logger.isDebugEnabled()) {
-			logger.debug("Scanning for failed item on write: " + inputs);
+			if (recovery) {
+				logger.debug("Scanning for failed item on recovery from write: "
+						+ inputs);
+			} else {
+				logger.debug("Scanning for failed item on write: " + inputs);
+			}
 		}
 		if (outputs.isEmpty()) {
 			inputs.setBusy(false);
@@ -527,15 +569,15 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 			contribution.incrementWriteCount(1);
 			inputIterator.remove();
 			outputIterator.remove();
-		}
-		catch (Exception e) {
+		} catch (Exception e) {
 			doOnWriteError(e, items);
-			if (!shouldSkip(itemWriteSkipPolicy, e, -1) && !rollbackClassifier.classify(e)) {
+			if (!shouldSkip(itemWriteSkipPolicy, e, -1)
+					&& !rollbackClassifier.classify(e)) {
 				inputIterator.remove();
 				outputIterator.remove();
-			}
-			else {
-				checkSkipPolicy(inputIterator, outputIterator, e, contribution);
+			} else {
+				checkSkipPolicy(inputIterator, outputIterator, e, contribution,
+						recovery);
 			}
 			if (rollbackClassifier.classify(e)) {
 				throw e;
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java
index e05010d4d..dac7a3d27 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java
@@ -17,10 +17,13 @@ import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.listener.ItemListenerSupport;
 import org.springframework.batch.core.step.skip.AlwaysSkipItemSkipPolicy;
+import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.item.support.PassThroughItemProcessor;
+import org.springframework.batch.retry.RetryException;
 import org.springframework.batch.retry.policy.NeverRetryPolicy;
+import org.springframework.batch.retry.policy.SimpleRetryPolicy;
 import org.springframework.dao.DataIntegrityViolationException;
 
 public class FaultTolerantChunkProcessorTests {
@@ -30,19 +33,22 @@ public class FaultTolerantChunkProcessorTests {
 	private List<String> list = new ArrayList<String>();
 
 	private List<String> after = new ArrayList<String>();
-	
+
 	private List<String> writeError = new ArrayList<String>();
 
 	private FaultTolerantChunkProcessor<String, String> processor;
 
-	private StepContribution contribution = new StepExecution("foo", new JobExecution(0L)).createStepContribution();
+	private StepContribution contribution = new StepExecution("foo",
+			new JobExecution(0L)).createStepContribution();
 
 	@Before
 	public void setUp() {
 		batchRetryTemplate = new BatchRetryTemplate();
-		processor = new FaultTolerantChunkProcessor<String, String>(new PassThroughItemProcessor<String>(),
+		processor = new FaultTolerantChunkProcessor<String, String>(
+				new PassThroughItemProcessor<String>(),
 				new ItemWriter<String>() {
-					public void write(List<? extends String> items) throws Exception {
+					public void write(List<? extends String> items)
+							throws Exception {
 						if (items.contains("fail")) {
 							throw new RuntimeException("Planned failure!");
 						}
@@ -90,8 +96,7 @@ public class FaultTolerantChunkProcessorTests {
 		try {
 			processor.process(contribution, inputs);
 			fail("Expected Exception");
-		}
-		catch (Exception e) {
+		} catch (Exception e) {
 			assertEquals("Skippable", e.getMessage());
 		}
 		processor.process(contribution, inputs);
@@ -102,6 +107,7 @@ public class FaultTolerantChunkProcessorTests {
 
 	/**
 	 * An Error can be retried or skipped but by default it is just propagated
+	 * 
 	 * @throws Exception
 	 */
 	@Test
@@ -114,12 +120,12 @@ public class FaultTolerantChunkProcessorTests {
 				}
 			}
 		});
-		Chunk<String> inputs = new Chunk<String>(Arrays.asList("3", "fail", "2"));
+		Chunk<String> inputs = new Chunk<String>(
+				Arrays.asList("3", "fail", "2"));
 		try {
 			processor.process(contribution, inputs);
 			fail("Expected Error");
-		}
-		catch (Error e) {
+		} catch (Error e) {
 			assertEquals("Expected Error!", e.getMessage());
 		}
 		processor.process(contribution, inputs);
@@ -135,20 +141,19 @@ public class FaultTolerantChunkProcessorTests {
 				}
 			}
 		});
-		Chunk<String> inputs = new Chunk<String>(Arrays.asList("3", "fail", "2"));
+		Chunk<String> inputs = new Chunk<String>(
+				Arrays.asList("3", "fail", "2"));
 		try {
 			processor.process(contribution, inputs);
 			fail("Expected RuntimeException");
-		}
-		catch (RuntimeException e) {
+		} catch (RuntimeException e) {
 			assertEquals("Expected Exception!", e.getMessage());
 		}
 		processor.process(contribution, inputs);
 		try {
 			processor.process(contribution, inputs);
 			fail("Expected RuntimeException");
-		}
-		catch (RuntimeException e) {
+		} catch (RuntimeException e) {
 			assertEquals("Expected Exception!", e.getMessage());
 		}
 		assertEquals(1, contribution.getSkipCount());
@@ -170,16 +175,15 @@ public class FaultTolerantChunkProcessorTests {
 		try {
 			processor.process(contribution, inputs);
 			fail("Expected RuntimeException");
-		}
-		catch (RuntimeException e) {
+		} catch (RuntimeException e) {
 			assertEquals("Expected Exception!", e.getMessage());
 		}
-		// BATCH-1518: ideally we would not want this to be necessary, but it still is...
+		// BATCH-1518: ideally we would not want this to be necessary, but it
+		// still is...
 		try {
 			processor.process(contribution, inputs);
 			fail("Expected RuntimeException");
-		}
-		catch (RuntimeException e) {
+		} catch (RuntimeException e) {
 			assertEquals("Expected Exception!", e.getMessage());
 		}
 		processor.process(contribution, inputs);
@@ -198,8 +202,11 @@ public class FaultTolerantChunkProcessorTests {
 			}
 		});
 		processor.setProcessSkipPolicy(new AlwaysSkipItemSkipPolicy());
-		processor.setRollbackClassifier(new BinaryExceptionClassifier(Collections
-				.<Class<? extends Throwable>> singleton(DataIntegrityViolationException.class), false));
+		processor
+				.setRollbackClassifier(new BinaryExceptionClassifier(
+						Collections
+								.<Class<? extends Throwable>> singleton(DataIntegrityViolationException.class),
+						false));
 		Chunk<String> inputs = new Chunk<String>(Arrays.asList("1", "2"));
 		processor.process(contribution, inputs);
 		assertEquals(1, list.size());
@@ -207,13 +214,15 @@ public class FaultTolerantChunkProcessorTests {
 
 	@Test
 	public void testAfterWrite() throws Exception {
-		Chunk<String> chunk = new Chunk<String>(Arrays.asList("foo", "fail", "bar"));
-		processor.setListeners(Arrays.asList(new ItemListenerSupport<String, String>() {
-			@Override
-			public void afterWrite(List<? extends String> item) {
-				after.addAll(item);
-			}
-		}));
+		Chunk<String> chunk = new Chunk<String>(Arrays.asList("foo", "fail",
+				"bar"));
+		processor.setListeners(Arrays
+				.asList(new ItemListenerSupport<String, String>() {
+					@Override
+					public void afterWrite(List<? extends String> item) {
+						after.addAll(item);
+					}
+				}));
 		processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy());
 		processAndExpectPlannedRuntimeException(chunk);
 		processor.process(contribution, chunk);
@@ -225,16 +234,19 @@ public class FaultTolerantChunkProcessorTests {
 		// foo is written once because it the failure is detected before it is
 		// committed the first time
 		assertEquals("[foo, bar]", list.toString());
-		// the after listener is called once per successful item, which is important
+		// the after listener is called once per successful item, which is
+		// important
 		assertEquals("[foo, bar]", after.toString());
 	}
 
 	@Test
 	public void testAfterWriteAllPassedInRecovery() throws Exception {
 		Chunk<String> chunk = new Chunk<String>(Arrays.asList("foo", "bar"));
-		processor = new FaultTolerantChunkProcessor<String, String>(new PassThroughItemProcessor<String>(),
+		processor = new FaultTolerantChunkProcessor<String, String>(
+				new PassThroughItemProcessor<String>(),
 				new ItemWriter<String>() {
-					public void write(List<? extends String> items) throws Exception {
+					public void write(List<? extends String> items)
+							throws Exception {
 						// Fail if there is more than one item
 						if (items.size() > 1) {
 							throw new RuntimeException("Planned failure!");
@@ -242,12 +254,13 @@ public class FaultTolerantChunkProcessorTests {
 						list.addAll(items);
 					}
 				}, batchRetryTemplate);
-		processor.setListeners(Arrays.asList(new ItemListenerSupport<String, String>() {
-			@Override
-			public void afterWrite(List<? extends String> item) {
-				after.addAll(item);
-			}
-		}));
+		processor.setListeners(Arrays
+				.asList(new ItemListenerSupport<String, String>() {
+					@Override
+					public void afterWrite(List<? extends String> item) {
+						after.addAll(item);
+					}
+				}));
 		processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy());
 
 		processAndExpectPlannedRuntimeException(chunk);
@@ -257,57 +270,216 @@ public class FaultTolerantChunkProcessorTests {
 		assertEquals("[foo, bar]", list.toString());
 		assertEquals("[foo, bar]", after.toString());
 	}
-	
+
 	@Test
-	public void testOnErrorInWrite() throws Exception{
+	public void testOnErrorInWrite() throws Exception {
 		Chunk<String> chunk = new Chunk<String>(Arrays.asList("foo", "fail"));
-		processor.setListeners(Arrays.asList(new ItemListenerSupport<String, String>() {
-			@Override
-			public void onWriteError(Exception e, List<? extends String> item) {
-				writeError.addAll(item);
-			}
-		}));
+		processor.setListeners(Arrays
+				.asList(new ItemListenerSupport<String, String>() {
+					@Override
+					public void onWriteError(Exception e,
+							List<? extends String> item) {
+						writeError.addAll(item);
+					}
+				}));
 		processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy());
-		
-		processAndExpectPlannedRuntimeException(chunk);//Process foo, fail		
-		processor.process(contribution, chunk);;//Process foo
-		processAndExpectPlannedRuntimeException(chunk);//Process fail
-		
+
+		processAndExpectPlannedRuntimeException(chunk);// Process foo, fail
+		processor.process(contribution, chunk);
+		;// Process foo
+		processAndExpectPlannedRuntimeException(chunk);// Process fail
+
 		assertEquals("[foo, fail, fail]", writeError.toString());
 	}
-	
+
 	@Test
-	public void testOnErrorInWriteAllItemsFail() throws Exception{
+	public void testOnErrorInWriteAllItemsFail() throws Exception {
 		Chunk<String> chunk = new Chunk<String>(Arrays.asList("foo", "bar"));
-		processor = new FaultTolerantChunkProcessor<String, String>(new PassThroughItemProcessor<String>(),
+		processor = new FaultTolerantChunkProcessor<String, String>(
+				new PassThroughItemProcessor<String>(),
 				new ItemWriter<String>() {
-					public void write(List<? extends String> items) throws Exception {
-						//Always fail in writer
+					public void write(List<? extends String> items)
+							throws Exception {
+						// Always fail in writer
 						throw new RuntimeException("Planned failure!");
 					}
 				}, batchRetryTemplate);
-		processor.setListeners(Arrays.asList(new ItemListenerSupport<String, String>() {
-			@Override
-			public void onWriteError(Exception e, List<? extends String> item) {
-				writeError.addAll(item);
-			}
-		}));
+		processor.setListeners(Arrays
+				.asList(new ItemListenerSupport<String, String>() {
+					@Override
+					public void onWriteError(Exception e,
+							List<? extends String> item) {
+						writeError.addAll(item);
+					}
+				}));
 		processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy());
-		
-		processAndExpectPlannedRuntimeException(chunk);//Process foo, bar		
-		processAndExpectPlannedRuntimeException(chunk);//Process foo
-		processAndExpectPlannedRuntimeException(chunk);//Process bar
-		
+
+		processAndExpectPlannedRuntimeException(chunk);// Process foo, bar
+		processAndExpectPlannedRuntimeException(chunk);// Process foo
+		processAndExpectPlannedRuntimeException(chunk);// Process bar
+
 		assertEquals("[foo, bar, foo, bar]", writeError.toString());
 	}
 
+	@Test
+	public void testWriteRetryOnException() throws Exception {
+		SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
+		retryPolicy.setMaxAttempts(2);
+		batchRetryTemplate.setRetryPolicy(retryPolicy);
+		processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy());
+		processor.setItemWriter(new ItemWriter<String>() {
+			public void write(List<? extends String> items) throws Exception {
+				if (items.contains("fail")) {
+					throw new IllegalArgumentException("Expected Exception!");
+				}
+			}
+		});
+		Chunk<String> inputs = new Chunk<String>(
+				Arrays.asList("3", "fail", "2"));
+		try {
+			processor.process(contribution, inputs);
+			fail("Expected RuntimeException");
+		} catch (RuntimeException e) {
+			assertEquals("Expected Exception!", e.getMessage());
+		}
+		try {
+			// first retry
+			processor.process(contribution, inputs);
+			fail("Expected RuntimeException");
+		} catch (RuntimeException e) {
+			assertEquals("Expected Exception!", e.getMessage());
+		}
+		// retry exhausted, now scanning
+		processor.process(contribution, inputs);
+		try {
+			// skip on this attempt
+			processor.process(contribution, inputs);
+			fail("Expected RuntimeException");
+		} catch (RuntimeException e) {
+			assertEquals("Expected Exception!", e.getMessage());
+		}
+		// finish chunk
+		processor.process(contribution, inputs);
+		assertEquals(1, contribution.getSkipCount());
+		assertEquals(2, contribution.getWriteCount());
+		assertEquals(0, contribution.getFilterCount());
+	}
+
+	@Test
+	public void testWriteRetryOnTwoExceptions() throws Exception {
+		SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
+		retryPolicy.setMaxAttempts(2);
+		batchRetryTemplate.setRetryPolicy(retryPolicy);
+		processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy());
+		processor.setItemWriter(new ItemWriter<String>() {
+			public void write(List<? extends String> items) throws Exception {
+				if (items.contains("fail")) {
+					throw new IllegalArgumentException("Expected Exception!");
+				}
+			}
+		});
+		Chunk<String> inputs = new Chunk<String>(Arrays.asList("3", "fail",
+				"fail", "4"));
+		try {
+			processor.process(contribution, inputs);
+			fail("Expected RuntimeException");
+		} catch (RuntimeException e) {
+			assertEquals("Expected Exception!", e.getMessage());
+		}
+		try {
+			// first retry
+			processor.process(contribution, inputs);
+			fail("Expected RuntimeException");
+		} catch (RuntimeException e) {
+			assertEquals("Expected Exception!", e.getMessage());
+		}
+		// retry exhausted, now scanning
+		processor.process(contribution, inputs);
+		try {
+			// skip on this attempt
+			processor.process(contribution, inputs);
+			fail("Expected RuntimeException");
+		} catch (RuntimeException e) {
+			assertEquals("Expected Exception!", e.getMessage());
+		}
+		try {
+			// 2nd exception detected
+			processor.process(contribution, inputs);
+			fail("Expected RuntimeException");
+		} catch (RuntimeException e) {
+			assertEquals("Expected Exception!", e.getMessage());
+		}
+		// still scanning
+		processor.process(contribution, inputs);
+		assertEquals(2, contribution.getSkipCount());
+		assertEquals(2, contribution.getWriteCount());
+		assertEquals(0, contribution.getFilterCount());
+	}
+
+	@Test
+	// BATCH-1804
+	public void testWriteRetryOnNonSkippableException() throws Exception {
+		SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
+		retryPolicy.setMaxAttempts(2);
+		batchRetryTemplate.setRetryPolicy(retryPolicy);
+		processor.setWriteSkipPolicy(new LimitCheckingItemSkipPolicy(1,
+				Collections.<Class<? extends Throwable>, Boolean> singletonMap(
+						IllegalArgumentException.class, true)));
+		processor.setItemWriter(new ItemWriter<String>() {
+			public void write(List<? extends String> items) throws Exception {
+				if (items.contains("fail")) {
+					throw new IllegalArgumentException("Expected Exception!");
+				}
+				if (items.contains("2")) {
+					throw new RuntimeException(
+							"Expected Non-Skippable Exception!");
+				}
+			}
+		});
+		Chunk<String> inputs = new Chunk<String>(
+				Arrays.asList("3", "fail", "2"));
+		try {
+			processor.process(contribution, inputs);
+			fail("Expected IllegalArgumentException");
+		} catch (IllegalArgumentException e) {
+			assertEquals("Expected Exception!", e.getMessage());
+		}
+		try {
+			// first retry
+			processor.process(contribution, inputs);
+			fail("Expected IllegalArgumentException");
+		} catch (IllegalArgumentException e) {
+			assertEquals("Expected Exception!", e.getMessage());
+		}
+		// retry exhausted, now scanning
+		processor.process(contribution, inputs);
+		try {
+			// skip on this attempt
+			processor.process(contribution, inputs);
+			fail("Expected IllegalArgumentException");
+		} catch (IllegalArgumentException e) {
+			assertEquals("Expected Exception!", e.getMessage());
+		}
+		try {
+			// should retry
+			processor.process(contribution, inputs);
+			fail("Expected RuntimeException");
+		} catch (RetryException e) {
+			throw e;
+		} catch (RuntimeException e) {
+			assertEquals("Expected Non-Skippable Exception!", e.getMessage());
+		}
+		assertEquals(1, contribution.getSkipCount());
+		assertEquals(1, contribution.getWriteCount());
+		assertEquals(0, contribution.getFilterCount());
+	}
+
 	protected void processAndExpectPlannedRuntimeException(Chunk<String> chunk)
 			throws Exception {
 		try {
 			processor.process(contribution, chunk);
 			fail();
-		}
-		catch (RuntimeException e) {
+		} catch (RuntimeException e) {
 			assertEquals("Planned failure!", e.getMessage());
 		}
 	}