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 new file mode 100644 index 000000000..384cc3bf0 --- /dev/null +++ b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepIntegrationTests.java @@ -0,0 +1,231 @@ +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; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.step.builder.FaultTolerantStepBuilder; +import org.springframework.batch.core.step.skip.SkipLimitExceededException; +import org.springframework.batch.core.step.skip.SkipPolicy; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.support.ListItemReader; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.test.context.ContextConfiguration; +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; + +/** + * Tests for fault tolerant {@link org.springframework.batch.core.step.item.ChunkOrientedTasklet}. + */ +@ContextConfiguration(locations = "/simple-job-launcher-context.xml") +@RunWith(SpringJUnit4ClassRunner.class) +public class FaultTolerantStepIntegrationTests { + + private static final int TOTAL_ITEMS = 30; + private static final int CHUNK_SIZE = TOTAL_ITEMS; + + @Autowired + private JobRepository jobRepository; + + @Autowired + private PlatformTransactionManager transactionManager; + + private SkipPolicy skipPolicy; + + private FaultTolerantStepBuilder stepBuilder; + + @Before + public void setUp() { + ItemReader itemReader = new ListItemReader<>(createItems()); + + ItemProcessor itemProcessor = new ItemProcessor() { + @Override + public Integer process(Integer item) throws Exception { + return item > 20 ? null : item; + } + }; + + ItemWriter itemWriter = new ItemWriter() { + @Override + public void write(List items) throws Exception { + if (items.contains(1)) { + throw new IllegalArgumentException(); + } + } + }; + + skipPolicy = new SkipIllegalArgumentExceptionSkipPolicy(); + stepBuilder = new StepBuilderFactory(jobRepository, transactionManager).get("step") + .chunk(CHUNK_SIZE) + .reader(itemReader) + .processor(itemProcessor) + .writer(itemWriter) + .faultTolerant(); + } + + @Test + public void testFilterCountWithTransactionalProcessorWhenSkipInWrite() throws Exception { + // Given + Step step = stepBuilder + .skipPolicy(skipPolicy) + .build(); + + // When + StepExecution stepExecution = execute(step); + + // Then + assertEquals(TOTAL_ITEMS, stepExecution.getReadCount()); + assertEquals(10, stepExecution.getFilterCount()); + assertEquals(19, stepExecution.getWriteCount()); + assertEquals(1, stepExecution.getWriteSkipCount()); + } + + @Test + public void testFilterCountWithNonTransactionalProcessorWhenSkipInWrite() throws Exception { + // Given + Step step = stepBuilder + .skipPolicy(skipPolicy) + .processorNonTransactional() + .build(); + + // When + StepExecution stepExecution = execute(step); + + // Then + assertEquals(TOTAL_ITEMS, stepExecution.getReadCount()); + assertEquals(10, stepExecution.getFilterCount()); + assertEquals(19, stepExecution.getWriteCount()); + assertEquals(1, stepExecution.getWriteSkipCount()); + } + + @Test + public void testFilterCountOnRetryWithTransactionalProcessorWhenSkipInWrite() throws Exception { + // Given + Step step = stepBuilder + .retry(IllegalArgumentException.class) + .retryLimit(2) + .skipPolicy(skipPolicy) + .build(); + + // When + StepExecution stepExecution = execute(step); + + // Then + assertEquals(TOTAL_ITEMS, stepExecution.getReadCount()); + // filter count is expected to be counted on each retry attempt + assertEquals(20, stepExecution.getFilterCount()); + assertEquals(19, stepExecution.getWriteCount()); + assertEquals(1, stepExecution.getWriteSkipCount()); + } + + @Test + public void testFilterCountOnRetryWithNonTransactionalProcessorWhenSkipInWrite() throws Exception { + // Given + Step step = stepBuilder + .retry(IllegalArgumentException.class) + .retryLimit(2) + .skipPolicy(skipPolicy) + .processorNonTransactional() + .build(); + + // When + StepExecution stepExecution = execute(step); + + // Then + assertEquals(TOTAL_ITEMS, stepExecution.getReadCount()); + // filter count is expected to be counted on each retry attempt + assertEquals(20, stepExecution.getFilterCount()); + 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++) { + items.add(i); + } + return items; + } + + private StepExecution execute(Step step) throws Exception { + JobExecution jobExecution = jobRepository.createJobExecution( + "job" + Math.random(), new JobParameters()); + StepExecution stepExecution = jobExecution.createStepExecution("step"); + jobRepository.add(stepExecution); + step.execute(stepExecution); + return stepExecution; + } + + private class SkipIllegalArgumentExceptionSkipPolicy implements SkipPolicy { + + @Override + public boolean shouldSkip(Throwable throwable, int skipCount) + throws SkipLimitExceededException { + return throwable instanceof IllegalArgumentException; + } + + } +} 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 4088277e7..0d641b0fa 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 @@ -16,8 +16,16 @@ package org.springframework.batch.core.step.item; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy; import org.springframework.batch.core.step.skip.NonSkippableProcessException; @@ -35,13 +43,6 @@ import org.springframework.retry.RetryContext; import org.springframework.retry.RetryException; import org.springframework.retry.support.DefaultRetryState; -import java.util.ArrayList; -import java.util.Collections; -import java.util.Iterator; -import java.util.List; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicReference; - /** * FaultTolerant implementation of the {@link ChunkProcessor} interface, that * allows for skipping or retry of items that cause exceptions during writing. @@ -572,6 +573,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 {