BATCH-2442: fix infinite loop when item processor fails during a scan

Currently, when the processor throws an exception during a scan, the
chunk is never marked as complete and the step never finishes. Moreover,
items that were processed unsuccessfully are still written.

This commit fixes the issue by excluding failed items from the scan.

Resolves BATCH-2442

Author: fmbenhassine

spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepIntegrationTests.java

@@ -3,6 +3,8 @@
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.ExitStatus;
 import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.Step;
@@ -22,6 +24,7 @@
 import org.springframework.transaction.PlatformTransactionManager;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 import static org.junit.Assert.assertEquals;
@@ -139,7 +142,55 @@ public void testFilterCountOnRetryWithNonTransactionalProcessorWhenSkipInWrite()
 		assertEquals(19, stepExecution.getWriteCount());
 		assertEquals(1, stepExecution.getWriteSkipCount());
 	}
-	
+
+	@Test(timeout = 3000)
+	public void testExceptionInProcessDuringChunkScan() throws Exception {
+		// Given
+		ListItemReader<Integer> itemReader = new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7));
+		ItemProcessor<Integer, Integer> itemProcessor = new ItemProcessor<Integer, Integer>() {
+			int cpt;
+
+			@Override
+			public Integer process(Integer item) throws Exception {
+				cpt++;
+				if (cpt == 7) { // item 2 succeeds the first time but fails during the scan
+					throw new Exception("Error during process");
+				}
+				return item;
+			}
+		};
+		ItemWriter<Integer> itemWriter = new ItemWriter<Integer>() {
+			int cpt;
+
+			@Override
+			public void write(List<? extends Integer> items) throws Exception {
+				cpt++;
+				if (cpt == 1) {
+					throw new Exception("Error during write");
+				}
+			}
+		};
+		Step step = new StepBuilderFactory(jobRepository, transactionManager).get("step")
+				.<Integer, Integer>chunk(5)
+				.reader(itemReader)
+				.processor(itemProcessor)
+				.writer(itemWriter)
+				.faultTolerant()
+				.skip(Exception.class)
+				.skipLimit(3)
+				.build();
+
+		// When
+		StepExecution stepExecution = execute(step);
+
+		// Then
+		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
+		assertEquals(ExitStatus.COMPLETED, stepExecution.getExitStatus());
+		assertEquals(7, stepExecution.getReadCount());
+		assertEquals(6, stepExecution.getWriteCount());
+		assertEquals(1, stepExecution.getProcessSkipCount());
+	}
+
 	private List<Integer> createItems() {
 		List<Integer> items = new ArrayList<>(TOTAL_ITEMS);
 		for (int i = 1; i <= TOTAL_ITEMS; i++) {

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

@@ -580,6 +580,14 @@ private void scan(final StepContribution contribution, final Chunk<I> inputs, fi
 		Chunk<I>.ChunkIterator inputIterator = inputs.iterator();
 		Chunk<O>.ChunkIterator outputIterator = outputs.iterator();
 
+		//BATCH-2442 : do not scan skipped items
+		if (!inputs.getSkips().isEmpty()) {
+			if (outputIterator.hasNext()) {
+				outputIterator.remove();
+				return;
+			}
+		}
+
 		List<O> items = Collections.singletonList(outputIterator.next());
 		inputIterator.next();
 		try {