Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

BATCH-2663: re-initialize filter count when scanning a chunk

Browse Source 
Before this commit, the filter count of the contribution was applied
for each item of a scanned chunk. For example, with a chunk of 30,
if the filter count is 10 and an item is skipped during write, then the
filter count is equal to 210 (10 + 20 * 10).

If this commit is applied, the filter count will be re-initialized when
scanning the chunk.

Resolves BATCH-2663
This commit is contained in:
Mahmoud Ben Hassine
2018-01-15 23:20:13 +01:00
committed by Michael Minella
parent d81ab76823
commit c78701a871
3 changed files with 231 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java
View File

@@ -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<String, String>() {
@Override
public String process(String item) throws Exception {
if (item.equals("1")) {
return null;
}
return item;
}
});
Chunk<String> inputs = new Chunk<String>(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<String, String>() {
@Override
public String process(String item) throws Exception {
if (item.equals("1")) {
return null;
}
return item;
}
});
Chunk<String> inputs = new Chunk<String>(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