From 346dcce78764985f5ff8cae1d1bc521f45f12a12 Mon Sep 17 00:00:00 2001 From: Mahmoud Ben Hassine 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 | 231 ++++++++++++++++++ .../item/FaultTolerantChunkProcessor.java | 23 +- 2 files changed, 247 insertions(+), 7 deletions(-) create mode 100644 spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepIntegrationTests.java 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 {