BATCH-1804: change scan behaviour if not in recovery
This commit is contained in:
Dave Syer
@@ -17,10 +17,13 @@ import org.springframework.batch.core.StepContribution;
|
||||
import org.springframework.batch.core.StepExecution;
|
||||
import org.springframework.batch.core.listener.ItemListenerSupport;
|
||||
import org.springframework.batch.core.step.skip.AlwaysSkipItemSkipPolicy;
|
||||
import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
|
||||
import org.springframework.batch.item.ItemProcessor;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
import org.springframework.batch.item.support.PassThroughItemProcessor;
|
||||
import org.springframework.batch.retry.RetryException;
|
||||
import org.springframework.batch.retry.policy.NeverRetryPolicy;
|
||||
import org.springframework.batch.retry.policy.SimpleRetryPolicy;
|
||||
import org.springframework.dao.DataIntegrityViolationException;
|
||||
|
||||
public class FaultTolerantChunkProcessorTests {
|
||||
@@ -30,19 +33,22 @@ public class FaultTolerantChunkProcessorTests {
|
||||
private List<String> list = new ArrayList<String>();
|
||||
|
||||
private List<String> after = new ArrayList<String>();
|
||||
|
||||
|
||||
private List<String> writeError = new ArrayList<String>();
|
||||
|
||||
private FaultTolerantChunkProcessor<String, String> processor;
|
||||
|
||||
private StepContribution contribution = new StepExecution("foo", new JobExecution(0L)).createStepContribution();
|
||||
private StepContribution contribution = new StepExecution("foo",
|
||||
new JobExecution(0L)).createStepContribution();
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
batchRetryTemplate = new BatchRetryTemplate();
|
||||
processor = new FaultTolerantChunkProcessor<String, String>(new PassThroughItemProcessor<String>(),
|
||||
processor = new FaultTolerantChunkProcessor<String, String>(
|
||||
new PassThroughItemProcessor<String>(),
|
||||
new ItemWriter<String>() {
|
||||
public void write(List<? extends String> items) throws Exception {
|
||||
public void write(List<? extends String> items)
|
||||
throws Exception {
|
||||
if (items.contains("fail")) {
|
||||
throw new RuntimeException("Planned failure!");
|
||||
}
|
||||
@@ -90,8 +96,7 @@ public class FaultTolerantChunkProcessorTests {
|
||||
try {
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected Exception");
|
||||
}
|
||||
catch (Exception e) {
|
||||
} catch (Exception e) {
|
||||
assertEquals("Skippable", e.getMessage());
|
||||
}
|
||||
processor.process(contribution, inputs);
|
||||
@@ -102,6 +107,7 @@ public class FaultTolerantChunkProcessorTests {
|
||||
|
||||
/**
|
||||
* An Error can be retried or skipped but by default it is just propagated
|
||||
*
|
||||
* @throws Exception
|
||||
*/
|
||||
@Test
|
||||
@@ -114,12 +120,12 @@ public class FaultTolerantChunkProcessorTests {
|
||||
}
|
||||
}
|
||||
});
|
||||
Chunk<String> inputs = new Chunk<String>(Arrays.asList("3", "fail", "2"));
|
||||
Chunk<String> inputs = new Chunk<String>(
|
||||
Arrays.asList("3", "fail", "2"));
|
||||
try {
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected Error");
|
||||
}
|
||||
catch (Error e) {
|
||||
} catch (Error e) {
|
||||
assertEquals("Expected Error!", e.getMessage());
|
||||
}
|
||||
processor.process(contribution, inputs);
|
||||
@@ -135,20 +141,19 @@ public class FaultTolerantChunkProcessorTests {
|
||||
}
|
||||
}
|
||||
});
|
||||
Chunk<String> inputs = new Chunk<String>(Arrays.asList("3", "fail", "2"));
|
||||
Chunk<String> inputs = new Chunk<String>(
|
||||
Arrays.asList("3", "fail", "2"));
|
||||
try {
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected RuntimeException");
|
||||
}
|
||||
catch (RuntimeException e) {
|
||||
} catch (RuntimeException e) {
|
||||
assertEquals("Expected Exception!", e.getMessage());
|
||||
}
|
||||
processor.process(contribution, inputs);
|
||||
try {
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected RuntimeException");
|
||||
}
|
||||
catch (RuntimeException e) {
|
||||
} catch (RuntimeException e) {
|
||||
assertEquals("Expected Exception!", e.getMessage());
|
||||
}
|
||||
assertEquals(1, contribution.getSkipCount());
|
||||
@@ -170,16 +175,15 @@ public class FaultTolerantChunkProcessorTests {
|
||||
try {
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected RuntimeException");
|
||||
}
|
||||
catch (RuntimeException e) {
|
||||
} catch (RuntimeException e) {
|
||||
assertEquals("Expected Exception!", e.getMessage());
|
||||
}
|
||||
// BATCH-1518: ideally we would not want this to be necessary, but it still is...
|
||||
// BATCH-1518: ideally we would not want this to be necessary, but it
|
||||
// still is...
|
||||
try {
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected RuntimeException");
|
||||
}
|
||||
catch (RuntimeException e) {
|
||||
} catch (RuntimeException e) {
|
||||
assertEquals("Expected Exception!", e.getMessage());
|
||||
}
|
||||
processor.process(contribution, inputs);
|
||||
@@ -198,8 +202,11 @@ public class FaultTolerantChunkProcessorTests {
|
||||
}
|
||||
});
|
||||
processor.setProcessSkipPolicy(new AlwaysSkipItemSkipPolicy());
|
||||
processor.setRollbackClassifier(new BinaryExceptionClassifier(Collections
|
||||
.<Class<? extends Throwable>> singleton(DataIntegrityViolationException.class), false));
|
||||
processor
|
||||
.setRollbackClassifier(new BinaryExceptionClassifier(
|
||||
Collections
|
||||
.<Class<? extends Throwable>> singleton(DataIntegrityViolationException.class),
|
||||
false));
|
||||
Chunk<String> inputs = new Chunk<String>(Arrays.asList("1", "2"));
|
||||
processor.process(contribution, inputs);
|
||||
assertEquals(1, list.size());
|
||||
@@ -207,13 +214,15 @@ public class FaultTolerantChunkProcessorTests {
|
||||
|
||||
@Test
|
||||
public void testAfterWrite() throws Exception {
|
||||
Chunk<String> chunk = new Chunk<String>(Arrays.asList("foo", "fail", "bar"));
|
||||
processor.setListeners(Arrays.asList(new ItemListenerSupport<String, String>() {
|
||||
@Override
|
||||
public void afterWrite(List<? extends String> item) {
|
||||
after.addAll(item);
|
||||
}
|
||||
}));
|
||||
Chunk<String> chunk = new Chunk<String>(Arrays.asList("foo", "fail",
|
||||
"bar"));
|
||||
processor.setListeners(Arrays
|
||||
.asList(new ItemListenerSupport<String, String>() {
|
||||
@Override
|
||||
public void afterWrite(List<? extends String> item) {
|
||||
after.addAll(item);
|
||||
}
|
||||
}));
|
||||
processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy());
|
||||
processAndExpectPlannedRuntimeException(chunk);
|
||||
processor.process(contribution, chunk);
|
||||
@@ -225,16 +234,19 @@ public class FaultTolerantChunkProcessorTests {
|
||||
// foo is written once because it the failure is detected before it is
|
||||
// committed the first time
|
||||
assertEquals("[foo, bar]", list.toString());
|
||||
// the after listener is called once per successful item, which is important
|
||||
// the after listener is called once per successful item, which is
|
||||
// important
|
||||
assertEquals("[foo, bar]", after.toString());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAfterWriteAllPassedInRecovery() throws Exception {
|
||||
Chunk<String> chunk = new Chunk<String>(Arrays.asList("foo", "bar"));
|
||||
processor = new FaultTolerantChunkProcessor<String, String>(new PassThroughItemProcessor<String>(),
|
||||
processor = new FaultTolerantChunkProcessor<String, String>(
|
||||
new PassThroughItemProcessor<String>(),
|
||||
new ItemWriter<String>() {
|
||||
public void write(List<? extends String> items) throws Exception {
|
||||
public void write(List<? extends String> items)
|
||||
throws Exception {
|
||||
// Fail if there is more than one item
|
||||
if (items.size() > 1) {
|
||||
throw new RuntimeException("Planned failure!");
|
||||
@@ -242,12 +254,13 @@ public class FaultTolerantChunkProcessorTests {
|
||||
list.addAll(items);
|
||||
}
|
||||
}, batchRetryTemplate);
|
||||
processor.setListeners(Arrays.asList(new ItemListenerSupport<String, String>() {
|
||||
@Override
|
||||
public void afterWrite(List<? extends String> item) {
|
||||
after.addAll(item);
|
||||
}
|
||||
}));
|
||||
processor.setListeners(Arrays
|
||||
.asList(new ItemListenerSupport<String, String>() {
|
||||
@Override
|
||||
public void afterWrite(List<? extends String> item) {
|
||||
after.addAll(item);
|
||||
}
|
||||
}));
|
||||
processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy());
|
||||
|
||||
processAndExpectPlannedRuntimeException(chunk);
|
||||
@@ -257,57 +270,216 @@ public class FaultTolerantChunkProcessorTests {
|
||||
assertEquals("[foo, bar]", list.toString());
|
||||
assertEquals("[foo, bar]", after.toString());
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testOnErrorInWrite() throws Exception{
|
||||
public void testOnErrorInWrite() throws Exception {
|
||||
Chunk<String> chunk = new Chunk<String>(Arrays.asList("foo", "fail"));
|
||||
processor.setListeners(Arrays.asList(new ItemListenerSupport<String, String>() {
|
||||
@Override
|
||||
public void onWriteError(Exception e, List<? extends String> item) {
|
||||
writeError.addAll(item);
|
||||
}
|
||||
}));
|
||||
processor.setListeners(Arrays
|
||||
.asList(new ItemListenerSupport<String, String>() {
|
||||
@Override
|
||||
public void onWriteError(Exception e,
|
||||
List<? extends String> item) {
|
||||
writeError.addAll(item);
|
||||
}
|
||||
}));
|
||||
processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy());
|
||||
|
||||
processAndExpectPlannedRuntimeException(chunk);//Process foo, fail
|
||||
processor.process(contribution, chunk);;//Process foo
|
||||
processAndExpectPlannedRuntimeException(chunk);//Process fail
|
||||
|
||||
|
||||
processAndExpectPlannedRuntimeException(chunk);// Process foo, fail
|
||||
processor.process(contribution, chunk);
|
||||
;// Process foo
|
||||
processAndExpectPlannedRuntimeException(chunk);// Process fail
|
||||
|
||||
assertEquals("[foo, fail, fail]", writeError.toString());
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testOnErrorInWriteAllItemsFail() throws Exception{
|
||||
public void testOnErrorInWriteAllItemsFail() throws Exception {
|
||||
Chunk<String> chunk = new Chunk<String>(Arrays.asList("foo", "bar"));
|
||||
processor = new FaultTolerantChunkProcessor<String, String>(new PassThroughItemProcessor<String>(),
|
||||
processor = new FaultTolerantChunkProcessor<String, String>(
|
||||
new PassThroughItemProcessor<String>(),
|
||||
new ItemWriter<String>() {
|
||||
public void write(List<? extends String> items) throws Exception {
|
||||
//Always fail in writer
|
||||
public void write(List<? extends String> items)
|
||||
throws Exception {
|
||||
// Always fail in writer
|
||||
throw new RuntimeException("Planned failure!");
|
||||
}
|
||||
}, batchRetryTemplate);
|
||||
processor.setListeners(Arrays.asList(new ItemListenerSupport<String, String>() {
|
||||
@Override
|
||||
public void onWriteError(Exception e, List<? extends String> item) {
|
||||
writeError.addAll(item);
|
||||
}
|
||||
}));
|
||||
processor.setListeners(Arrays
|
||||
.asList(new ItemListenerSupport<String, String>() {
|
||||
@Override
|
||||
public void onWriteError(Exception e,
|
||||
List<? extends String> item) {
|
||||
writeError.addAll(item);
|
||||
}
|
||||
}));
|
||||
processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy());
|
||||
|
||||
processAndExpectPlannedRuntimeException(chunk);//Process foo, bar
|
||||
processAndExpectPlannedRuntimeException(chunk);//Process foo
|
||||
processAndExpectPlannedRuntimeException(chunk);//Process bar
|
||||
|
||||
|
||||
processAndExpectPlannedRuntimeException(chunk);// Process foo, bar
|
||||
processAndExpectPlannedRuntimeException(chunk);// Process foo
|
||||
processAndExpectPlannedRuntimeException(chunk);// Process bar
|
||||
|
||||
assertEquals("[foo, bar, foo, bar]", writeError.toString());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWriteRetryOnException() throws Exception {
|
||||
SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
|
||||
retryPolicy.setMaxAttempts(2);
|
||||
batchRetryTemplate.setRetryPolicy(retryPolicy);
|
||||
processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy());
|
||||
processor.setItemWriter(new ItemWriter<String>() {
|
||||
public void write(List<? extends String> items) throws Exception {
|
||||
if (items.contains("fail")) {
|
||||
throw new IllegalArgumentException("Expected Exception!");
|
||||
}
|
||||
}
|
||||
});
|
||||
Chunk<String> inputs = new Chunk<String>(
|
||||
Arrays.asList("3", "fail", "2"));
|
||||
try {
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected RuntimeException");
|
||||
} catch (RuntimeException e) {
|
||||
assertEquals("Expected Exception!", e.getMessage());
|
||||
}
|
||||
try {
|
||||
// first retry
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected RuntimeException");
|
||||
} catch (RuntimeException e) {
|
||||
assertEquals("Expected Exception!", e.getMessage());
|
||||
}
|
||||
// retry exhausted, now scanning
|
||||
processor.process(contribution, inputs);
|
||||
try {
|
||||
// skip on this attempt
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected RuntimeException");
|
||||
} catch (RuntimeException e) {
|
||||
assertEquals("Expected Exception!", e.getMessage());
|
||||
}
|
||||
// finish chunk
|
||||
processor.process(contribution, inputs);
|
||||
assertEquals(1, contribution.getSkipCount());
|
||||
assertEquals(2, contribution.getWriteCount());
|
||||
assertEquals(0, contribution.getFilterCount());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWriteRetryOnTwoExceptions() throws Exception {
|
||||
SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
|
||||
retryPolicy.setMaxAttempts(2);
|
||||
batchRetryTemplate.setRetryPolicy(retryPolicy);
|
||||
processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy());
|
||||
processor.setItemWriter(new ItemWriter<String>() {
|
||||
public void write(List<? extends String> items) throws Exception {
|
||||
if (items.contains("fail")) {
|
||||
throw new IllegalArgumentException("Expected Exception!");
|
||||
}
|
||||
}
|
||||
});
|
||||
Chunk<String> inputs = new Chunk<String>(Arrays.asList("3", "fail",
|
||||
"fail", "4"));
|
||||
try {
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected RuntimeException");
|
||||
} catch (RuntimeException e) {
|
||||
assertEquals("Expected Exception!", e.getMessage());
|
||||
}
|
||||
try {
|
||||
// first retry
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected RuntimeException");
|
||||
} catch (RuntimeException e) {
|
||||
assertEquals("Expected Exception!", e.getMessage());
|
||||
}
|
||||
// retry exhausted, now scanning
|
||||
processor.process(contribution, inputs);
|
||||
try {
|
||||
// skip on this attempt
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected RuntimeException");
|
||||
} catch (RuntimeException e) {
|
||||
assertEquals("Expected Exception!", e.getMessage());
|
||||
}
|
||||
try {
|
||||
// 2nd exception detected
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected RuntimeException");
|
||||
} catch (RuntimeException e) {
|
||||
assertEquals("Expected Exception!", e.getMessage());
|
||||
}
|
||||
// still scanning
|
||||
processor.process(contribution, inputs);
|
||||
assertEquals(2, contribution.getSkipCount());
|
||||
assertEquals(2, contribution.getWriteCount());
|
||||
assertEquals(0, contribution.getFilterCount());
|
||||
}
|
||||
|
||||
@Test
|
||||
// BATCH-1804
|
||||
public void testWriteRetryOnNonSkippableException() throws Exception {
|
||||
SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
|
||||
retryPolicy.setMaxAttempts(2);
|
||||
batchRetryTemplate.setRetryPolicy(retryPolicy);
|
||||
processor.setWriteSkipPolicy(new LimitCheckingItemSkipPolicy(1,
|
||||
Collections.<Class<? extends Throwable>, Boolean> singletonMap(
|
||||
IllegalArgumentException.class, true)));
|
||||
processor.setItemWriter(new ItemWriter<String>() {
|
||||
public void write(List<? extends String> items) throws Exception {
|
||||
if (items.contains("fail")) {
|
||||
throw new IllegalArgumentException("Expected Exception!");
|
||||
}
|
||||
if (items.contains("2")) {
|
||||
throw new RuntimeException(
|
||||
"Expected Non-Skippable Exception!");
|
||||
}
|
||||
}
|
||||
});
|
||||
Chunk<String> inputs = new Chunk<String>(
|
||||
Arrays.asList("3", "fail", "2"));
|
||||
try {
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected IllegalArgumentException");
|
||||
} catch (IllegalArgumentException e) {
|
||||
assertEquals("Expected Exception!", e.getMessage());
|
||||
}
|
||||
try {
|
||||
// first retry
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected IllegalArgumentException");
|
||||
} catch (IllegalArgumentException e) {
|
||||
assertEquals("Expected Exception!", e.getMessage());
|
||||
}
|
||||
// retry exhausted, now scanning
|
||||
processor.process(contribution, inputs);
|
||||
try {
|
||||
// skip on this attempt
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected IllegalArgumentException");
|
||||
} catch (IllegalArgumentException e) {
|
||||
assertEquals("Expected Exception!", e.getMessage());
|
||||
}
|
||||
try {
|
||||
// should retry
|
||||
processor.process(contribution, inputs);
|
||||
fail("Expected RuntimeException");
|
||||
} catch (RetryException e) {
|
||||
throw e;
|
||||
} catch (RuntimeException e) {
|
||||
assertEquals("Expected Non-Skippable Exception!", e.getMessage());
|
||||
}
|
||||
assertEquals(1, contribution.getSkipCount());
|
||||
assertEquals(1, contribution.getWriteCount());
|
||||
assertEquals(0, contribution.getFilterCount());
|
||||
}
|
||||
|
||||
protected void processAndExpectPlannedRuntimeException(Chunk<String> chunk)
|
||||
throws Exception {
|
||||
try {
|
||||
processor.process(contribution, chunk);
|
||||
fail();
|
||||
}
|
||||
catch (RuntimeException e) {
|
||||
} catch (RuntimeException e) {
|
||||
assertEquals("Planned failure!", e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user