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 itemReader = new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7)); + ItemProcessor itemProcessor = new ItemProcessor() { + 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 itemWriter = new ItemWriter() { + int cpt; + + @Override + public void write(List items) throws Exception { + cpt++; + if (cpt == 1) { + throw new Exception("Error during write"); + } + } + }; + Step step = new StepBuilderFactory(jobRepository, transactionManager).get("step") + .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 createItems() { List 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 extends SimpleChunkProcessor.ChunkIterator inputIterator = inputs.iterator(); Chunk.ChunkIterator outputIterator = outputs.iterator(); + //BATCH-2442 : do not scan skipped items + if (!inputs.getSkips().isEmpty()) { + if (outputIterator.hasNext()) { + outputIterator.remove(); + return; + } + } + List items = Collections.singletonList(outputIterator.next()); inputIterator.next(); try {