From c1b79e5946e6bd2fa9b194833bdcb6917579dcd5 Mon Sep 17 00:00:00 2001
From: Mahmoud Ben Hassine <mbenhassine@pivotal.io>
Date: Tue, 27 Mar 2018 14:07:27 +0200
Subject: [PATCH] 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
---
 .../FaultTolerantStepIntegrationTests.java    | 53 ++++++++++++++++++-
 .../item/FaultTolerantChunkProcessor.java     |  8 +++
 2 files changed, 60 insertions(+), 1 deletion(-)

diff --git a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepIntegrationTests.java b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepIntegrationTests.java
index a1fc034d6..2dae35046 100644
--- a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepIntegrationTests.java
+++ b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepIntegrationTests.java
@@ -3,6 +3,8 @@ package org.springframework.batch.core.test.step;
 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.test.context.junit4.SpringJUnit4ClassRunner;
 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 class FaultTolerantStepIntegrationTests {
 		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++) {
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 a7746e662..83a13ec6b 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
@@ -580,6 +580,14 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 		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 {