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..a1fc034d6 --- /dev/null +++ b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepIntegrationTests.java @@ -0,0 +1,169 @@ +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.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.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 = item -> item > 20 ? null : item; + ItemWriter itemWriter = chunk -> { + if (chunk.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()); + } + + 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 30c9f7039..a7746e662 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 @@ -1,5 +1,5 @@ /* - * Copyright 2006-2013 the original author or authors. + * Copyright 2006-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -149,6 +149,12 @@ public class FaultTolerantChunkProcessor extends SimpleChunkProcessor()); } + else { + // BATCH-2663: re-initialize filter count when scanning the chunk + if (data.scanning()) { + data.filterCount = 0; + } + } } @Override diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java index 251931183..8a5ef03b9 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java @@ -122,6 +122,61 @@ public class FaultTolerantChunkProcessorTests { assertEquals(1, contribution.getSkipCount()); assertEquals(1, contribution.getFilterCount()); } + + @Test + // BATCH-2663 + public void testFilterCountOnSkipInWriteWithoutRetry() throws Exception { + processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy()); + processor.setItemProcessor(new ItemProcessor() { + @Override + public String process(String item) throws Exception { + if (item.equals("1")) { + return null; + } + return item; + } + }); + Chunk inputs = new Chunk(Arrays.asList("fail", "1", "2")); + processAndExpectPlannedRuntimeException(inputs); // (first attempt) Process fail, 1, 2 + // item 1 is filtered out so it is removed from the chunk => now inputs = [fail, 2] + // using NeverRetryPolicy by default => now scanning + processAndExpectPlannedRuntimeException(inputs); // (scanning) Process fail + processor.process(contribution, inputs); // (scanning) Process 2 + assertEquals(1, list.size()); + assertEquals("[2]", list.toString()); + assertEquals(1, contribution.getWriteSkipCount()); + assertEquals(1, contribution.getFilterCount()); + } + + @Test + // BATCH-2663 + public void testFilterCountOnSkipInWriteWithRetry() throws Exception { + SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(); + retryPolicy.setMaxAttempts(3); + batchRetryTemplate.setRetryPolicy(retryPolicy); + processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy()); + processor.setItemProcessor(new ItemProcessor() { + @Override + public String process(String item) throws Exception { + if (item.equals("1")) { + return null; + } + return item; + } + }); + Chunk inputs = new Chunk(Arrays.asList("fail", "1", "2")); + processAndExpectPlannedRuntimeException(inputs); // (first attempt) Process fail, 1, 2 + // item 1 is filtered out so it is removed from the chunk => now inputs = [fail, 2] + processAndExpectPlannedRuntimeException(inputs); // (first retry) Process fail, 2 + processAndExpectPlannedRuntimeException(inputs); // (second retry) Process fail, 2 + // retry exhausted (maxAttempts = 3) => now scanning + processAndExpectPlannedRuntimeException(inputs); // (scanning) Process fail + processor.process(contribution, inputs); // (scanning) Process 2 + assertEquals(1, list.size()); + assertEquals("[2]", list.toString()); + assertEquals(1, contribution.getWriteSkipCount()); + assertEquals(3, contribution.getFilterCount()); + } /** * An Error can be retried or skipped but by default it is just propagated