OPEN - BATCH-847: FaultTolerantChunkOrientedTasklet loses chunks when non-skippable exceptions thrown in read phase

exception on read is rethrown as fatal, added logging for exceptions swallowed in SimpleRetryExceptionHandler

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTasklet.java

@@ -129,10 +129,11 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
 			process(contribution, inputs, outputs);
 		}
 
-		/* Savepoint at end of processing and before writing. The processed items
-		 * ready for output are stored so that if writing fails they can be picked
-		 * up again in the next try. The inputs are finished with so we can clear
-		 * their attribute.
+		/*
+		 * Savepoint at end of processing and before writing. The processed
+		 * items ready for output are stored so that if writing fails they can
+		 * be picked up again in the next try. The inputs are finished with so
+		 * we can clear their attribute.
 		 */
 		attributes.setAttribute(OUTPUT_BUFFER_KEY, outputs);
 		attributes.removeAttribute(INPUT_BUFFER_KEY);
@@ -167,27 +168,23 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
 				return doRead();
 			}
 			catch (Exception e) {
-				try {
-					if (readSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
-						// increment skip count and try again
-						contribution.incrementReadSkipCount();
-						try {
-							listener.onSkipInRead(e);
-						}
-						catch (RuntimeException ex) {
-							throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
-						}
-						logger.debug("Skipping failed input", e);
+
+				if (readSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
+					// increment skip count and try again
+					contribution.incrementReadSkipCount();
+					try {
+						listener.onSkipInRead(e);
 					}
-					else {
-						// skip doesn't apply -> rethrow
-						throw e;
+					catch (RuntimeException ex) {
+						throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
 					}
+					logger.debug("Skipping failed input", e);
 				}
-				catch (SkipLimitExceededException ex) {
-					// re-throw when the skip policy runs out of patience
-					throw ex;
+				else {
+					// skip doesn't apply -> rethrow as if skip limit of zero was exceeded
+					throw new SkipLimitExceededException(0, e);
 				}
+
 			}
 		}
 

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleRetryExceptionHandler.java

@@ -17,6 +17,8 @@ package org.springframework.batch.core.step.item;
 
 import java.util.Collection;
 
+import org.apache.commons.logging.LogFactory;
+import org.apache.commons.logging.Log;
 import org.springframework.batch.repeat.RepeatContext;
 import org.springframework.batch.repeat.exception.ExceptionHandler;
 import org.springframework.batch.repeat.support.RepeatSynchronizationManager;
@@ -40,6 +42,8 @@ public class SimpleRetryExceptionHandler extends RetryListenerSupport implements
 	 * Attribute key, whose existence signals an exhausted retry.
 	 */
 	private static final String EXHAUSTED = SimpleRetryExceptionHandler.class.getName() + ".RETRY_EXHAUSTED";
+	
+	private static final Log logger = LogFactory.getLog(SimpleRetryExceptionHandler.class);
 
 	final private RetryPolicy retryPolicy;
 
@@ -77,6 +81,9 @@ public class SimpleRetryExceptionHandler extends RetryListenerSupport implements
 		if (fatalExceptionClassifier.classify(throwable) || context.hasAttribute(EXHAUSTED)) {
 			exceptionHandler.handleException(context, throwable);
 		}
+		else {
+			logger.debug("handled non-fatal exception", throwable);
+		}
 	}
 
 	/**

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java

@@ -72,6 +72,58 @@ public class SkipLimitStepFactoryBeanTests {
 		jobExecution = new JobExecution(jobInstance);
 	}
 
+	/**
+	 * Non-skippable (and non-fatal) exception causes failure immediately.
+	 * @throws Exception
+	 */
+	@Test
+	public void testNonSkippableExceptionOnRead() throws Exception {
+		
+		// nothing is skippable
+		Collection<Class<? extends Throwable>> empty = Collections.emptySet();
+		factory.setSkippableExceptionClasses(empty);
+		
+		// no exceptions on write
+		factory.setItemWriter(new ItemWriter<String>() {
+			public void write(List<? extends String> items) throws Exception {
+				logger.debug(items);
+			}
+		});
+
+		Step step = (Step) factory.getObject();
+		StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
+
+		step.execute(stepExecution);
+		assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
+		//assertEquals("Ouch!", stepExecution.getFailureExceptions().get(0).getMessage());
+	}
+	
+	@Test
+	public void testNonSkippableException() throws Exception {
+		// nothing is skippable
+		Collection<Class<? extends Throwable>> empty = Collections.emptySet();
+		factory.setSkippableExceptionClasses(empty);
+		factory.setCommitInterval(1);
+		
+		// no failures on read
+		reader = new SkipReaderStub(new String[] { "1", "2", "3", "4", "5" }, new ArrayList<String>());
+		factory.setItemReader(reader);
+		factory.setItemWriter(new ItemWriter<String>() {
+
+			public void write(List<? extends String> items) throws Exception {
+				throw new RuntimeException("non-skippable exception");
+			}
+			
+		});
+
+		Step step = (Step) factory.getObject();
+		StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
+
+		step.execute(stepExecution);
+		assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
+		assertEquals(1, reader.processed.size());
+	}
+
 	/**
 	 * Check items causing errors are skipped as expected.
 	 */
@@ -99,6 +151,7 @@ public class SkipLimitStepFactoryBeanTests {
 		assertEquals(expectedOutput, writer.written);
 
 		assertEquals(4, stepExecution.getReadCount());
+		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
 
 	}
 
@@ -195,7 +248,7 @@ public class SkipLimitStepFactoryBeanTests {
 
 		step.execute(stepExecution);
 		assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
-		
+
 		assertEquals(3, stepExecution.getSkipCount());
 		assertEquals(2, stepExecution.getReadSkipCount());
 		assertEquals(1, stepExecution.getWriteSkipCount());

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java

@@ -159,13 +159,13 @@ public class StatefulRetryStepFactoryBeanTests {
 
 		assertEquals(0, stepExecution.getSkipCount());
 
-		// [a, b, c, null]
-		assertEquals(4, provided.size());
-		// [a, c]
-		assertEquals(2, processed.size());
+		// [a, b with error]
+		assertEquals(2, provided.size());
+		// [a]
+		assertEquals(1, processed.size());
 		// []
 		assertEquals(0, recovered.size());
-		assertEquals(2, stepExecution.getReadCount());
+		assertEquals(1, stepExecution.getReadCount());
 		assertEquals(0, stepExecution.getReadSkipCount());
 	}