RESOLVED - issue BATCH-1518: Add optimization for commit-interval=1
This commit is contained in:
@@ -208,7 +208,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
|
||||
O output = null;
|
||||
try {
|
||||
count.incrementAndGet();
|
||||
O cached = (cacheIterator != null&& cacheIterator.hasNext()) ? cacheIterator.next() : null;
|
||||
O cached = (cacheIterator != null && cacheIterator.hasNext()) ? cacheIterator.next() : null;
|
||||
if (cached != null && count.get() > scanLimit) {
|
||||
/*
|
||||
* If there is a cached chunk then we must be
|
||||
@@ -221,7 +221,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
|
||||
}
|
||||
else {
|
||||
output = doProcess(item);
|
||||
if (!processorTransactional) {
|
||||
if (!processorTransactional) {
|
||||
cache.add(output);
|
||||
}
|
||||
}
|
||||
@@ -377,6 +377,12 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
|
||||
"Retry exhausted after last attempt in recovery path, but exception is not skippable.",
|
||||
context.getLastThrowable());
|
||||
}
|
||||
|
||||
if (inputs.size()==1 && outputs.getSkips().isEmpty()) {
|
||||
// Optimization for trivial chunk size: no need to scan
|
||||
checkSkipPolicy(inputs.iterator(), outputs.iterator(), context.getLastThrowable(), contribution);
|
||||
return null;
|
||||
}
|
||||
|
||||
inputs.setBusy(true);
|
||||
scan(contribution, inputs, outputs, chunkMonitor);
|
||||
|
||||
@@ -154,6 +154,30 @@ public class FaultTolerantChunkProcessorTests {
|
||||
assertEquals(0, contribution.getFilterCount());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWriteSkipOnExceptionWithTrivialChunk() throws Exception {
|
||||
processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy());
|
||||
processor.setItemWriter(new ItemWriter<String>() {
|
||||
public void write(List<? extends String> items) throws Exception {
|
||||
if (items.contains("fail")) {
|
||||
throw new RuntimeException("Expected Exception!");
|
||||
}
|
||||
}
|
||||
});
|
||||
Chunk<String> inputs = new Chunk<String>(Arrays.asList("fail"));
|
||||
try {
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected RuntimeException");
|
||||
}
|
||||
catch (RuntimeException e) {
|
||||
assertEquals("Expected Exception!", e.getMessage());
|
||||
}
|
||||
processor.process(contribution, inputs);
|
||||
assertEquals(1, contribution.getSkipCount());
|
||||
assertEquals(0, contribution.getWriteCount());
|
||||
assertEquals(0, contribution.getFilterCount());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testTransformWithExceptionAndNoRollback() throws Exception {
|
||||
processor.setItemProcessor(new ItemProcessor<String, String>() {
|
||||
@@ -200,11 +224,11 @@ public class FaultTolerantChunkProcessorTests {
|
||||
assertEquals(1, chunk.getItems().size());
|
||||
processor.process(contribution, chunk);
|
||||
assertEquals(0, chunk.getItems().size());
|
||||
// foo is written twice because the failure is detected on the second
|
||||
// attempt when throttling
|
||||
// foo is written once because it the failure is detected before it is
|
||||
// committed the first time
|
||||
assertEquals("[foo, bar]", list.toString());
|
||||
// but the after listener is only called once, which is important
|
||||
assertEquals(2, after.size());
|
||||
// the after listener is called once per successful item, which is important
|
||||
assertEquals("[foo, bar]", after.toString());
|
||||
}
|
||||
|
||||
@Test
|
||||
|
||||
@@ -304,7 +304,7 @@ public class FaultTolerantStepFactoryBeanRetryTests {
|
||||
assertEquals(expectedOutput, written);
|
||||
|
||||
assertEquals("[a, b, c, d, e, f, null]", provided.toString());
|
||||
assertEquals("[a, b, b, b, b, b, b, c, d, d, d, d, d, d, e, f]", processed.toString());
|
||||
assertEquals("[a, b, b, b, b, b, c, d, d, d, d, d, e, f]", processed.toString());
|
||||
assertEquals("[b, d]", recovered.toString());
|
||||
}
|
||||
|
||||
@@ -361,7 +361,7 @@ public class FaultTolerantStepFactoryBeanRetryTests {
|
||||
assertEquals(7, provided.size());
|
||||
// [a, b, c, a, b, c, a, b, c, a, b, c, a, b, c, a, b, c, d, e, f, d,
|
||||
// e, f, d, e, f, d, e, f, d, e, f, d, e, f]
|
||||
System.err.println(processed);
|
||||
// System.err.println(processed);
|
||||
assertEquals(36, processed.size());
|
||||
// [b, d]
|
||||
assertEquals(2, recovered.size());
|
||||
@@ -405,7 +405,7 @@ public class FaultTolerantStepFactoryBeanRetryTests {
|
||||
assertEquals(1, provided.size());
|
||||
// the failed items are tried up to the limit (but only precisely so if
|
||||
// the commit interval is 1)
|
||||
assertEquals("[b, b, b, b, b]", processed.toString());
|
||||
assertEquals("[b, b, b, b]", processed.toString());
|
||||
// []
|
||||
assertEquals(0, recovered.size());
|
||||
assertEquals(1, stepExecution.getReadCount());
|
||||
@@ -496,7 +496,7 @@ public class FaultTolerantStepFactoryBeanRetryTests {
|
||||
assertEquals(0, stepExecution.getSkipCount());
|
||||
// [b]
|
||||
assertEquals(1, provided.size());
|
||||
assertEquals("[b, b, b, b, b]", processed.toString());
|
||||
assertEquals("[b, b, b, b]", processed.toString());
|
||||
// []
|
||||
assertEquals(0, recovered.size());
|
||||
assertEquals(1, stepExecution.getReadCount());
|
||||
|
||||
Reference in New Issue
Block a user