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

RESOLVED - issue BATCH-1354: Infinite loop caused by throwing an Error from the ItemWriter of a skippable step

Browse Source 
http://jira.springframework.org/browse/BATCH-1354
This commit is contained in:
dsyer
2009-08-20 15:43:19 +00:00
parent 4581617d18
commit a293b6ff21
2 changed files with 83 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
View File

@@ -283,8 +283,11 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
try {
doWrite(outputs.getItems());
}
catch (Exception e) {
if (rollbackClassifier.classify(e)) {
catch (Throwable t) {
Exception e = (t instanceof Exception) ? (Exception) t : new IllegalStateException(
"Unexpected non-Exception Throwable", t);
if (rollbackClassifier.classify(t)) {
throw e;
}
/*
@@ -312,7 +315,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
public Object recover(RetryContext context) throws Exception {
Exception e = (Exception) context.getLastThrowable();
Exception e = context.getLastThrowable();
if (outputs.size() > 1 && !rollbackClassifier.classify(e)) {
throw new RetryException("Invalid retry state during write caused by "
+ "exception that does not classify for rollback: ", e);
@@ -384,7 +387,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
if (item == null) {
continue;
}
Exception e = wrapper.getException();
Throwable e = wrapper.getException();
try {
getListener().onSkipInProcess(item, e);
}
@@ -394,7 +397,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
}
for (SkipWrapper<O> wrapper : outputs.getSkips()) {
Exception e = wrapper.getException();
Throwable e = wrapper.getException();
try {
getListener().onSkipInWrite(wrapper.getItem(), e);
}
@@ -464,16 +467,20 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
inputIterator.remove();
outputIterator.remove();
}
catch (Exception e) {
if (!itemWriteSkipPolicy.shouldSkip(e, -1) && !rollbackClassifier.classify(e)) {
catch (Throwable t) {
Exception e = (t instanceof Exception) ? (Exception) t : new IllegalStateException(
"Unexpected non-Exception Throwable", t);
if (!itemWriteSkipPolicy.shouldSkip(e, -1) && !rollbackClassifier.classify(t)) {
inputIterator.remove();
outputIterator.remove();
}
else {
checkSkipPolicy(inputIterator, outputIterator, e, contribution);
}
if (rollbackClassifier.classify(e)) {
throw e;
if (rollbackClassifier.classify(t)) {
throw e;
}
}
chunkMonitor.incrementOffset();

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

@@ -1,6 +1,7 @@
package org.springframework.batch.core.step.item;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.fail;
import java.util.ArrayList;
@@ -87,7 +88,8 @@ 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);
@@ -96,16 +98,78 @@ public class FaultTolerantChunkProcessorTests {
assertEquals(1, contribution.getFilterCount());
}
@Test
public void testWriteSkipOnError() throws Exception {
processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy());
processor.setItemWriter(new ItemWriter<String>() {
public void write(List<? extends String> items) throws Exception {
if (items.contains("fail")) {
assertFalse("Expected Error!", true);
}
}
});
Chunk<String> inputs = new Chunk<String>(Arrays.asList("3", "fail", "2"));
try {
processor.process(contribution, inputs);
fail("Expected Error");
}
catch (IllegalStateException e) {
assertEquals("Expected Error!", e.getCause().getMessage());
}
processor.process(contribution, inputs);
try {
processor.process(contribution, inputs);
fail("Expected Error");
}
catch (IllegalStateException e) {
assertEquals("Expected Error!", e.getCause().getMessage());
}
assertEquals(1, contribution.getSkipCount());
assertEquals(1, contribution.getWriteCount());
}
@Test
public void testWriteSkipOnException() 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("3", "fail", "2"));
try {
processor.process(contribution, inputs);
fail("Expected RuntimeException");
}
catch (RuntimeException e) {
assertEquals("Expected Exception!", e.getMessage());
}
processor.process(contribution, inputs);
try {
processor.process(contribution, inputs);
fail("Expected RuntimeException");
}
catch (RuntimeException e) {
assertEquals("Expected Exception!", e.getMessage());
}
assertEquals(1, contribution.getSkipCount());
assertEquals(1, contribution.getWriteCount());
}
@Test
public void testTransformWithExceptionAndNoRollback() throws Exception {
processor.setItemProcessor(new ItemProcessor<String, String>() {
public String process(String item) throws Exception {
if (item.equals("1")) throw new DataIntegrityViolationException("Planned");
if (item.equals("1"))
throw new DataIntegrityViolationException("Planned");
return item;
}
});
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());