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

RESOLVED - BATCH-887: onSkipInProcess called multiple times for same item using FaultTolerantChunkOrientedTasklet

Browse Source
This commit is contained in:
robokaso
2008-10-24 08:19:41 +00:00
parent b19ea10f2f
commit 2acb661331
2 changed files with 97 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTasklet.java
View File

@@ -224,6 +224,12 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
if (processSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
contribution.incrementProcessSkipCount();
iterator.remove(e);
try {
listener.onSkipInProcess(item, e);
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
}
return null;
}
else {
@@ -244,16 +250,6 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
}
for (ItemWrapper<T> skip : inputs.getSkips()) {
Exception exception = skip.getException();
try {
listener.onSkipInProcess(skip.getItem(), exception);
}
catch (RuntimeException e) {
throw new SkipListenerFailedException("Fatal exception in SkipListener.", e, exception);
}
}
contribution.incrementFilterCount(filtered);
}
@@ -287,25 +283,25 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
return null;
}
for (S item : chunk) {
try {
doWrite(Collections.singletonList(item));
contribution.incrementWriteCount(1);
try {
doWrite(Collections.singletonList(item));
contribution.incrementWriteCount(1);
}
catch (Exception e) {
checkSkipPolicy(item, e, contribution);
if (rollbackClassifier.classify(e)) {
throw e;
}
catch (Exception e) {
checkSkipPolicy(item, e, contribution);
if (rollbackClassifier.classify(e)) {
throw e;
}
else {
logger.error("Exception encountered that does not classify for rollback: ", e);
}
else {
logger.error("Exception encountered that does not classify for rollback: ", e);
}
}
}
return null;
}
private void checkSkipPolicy(S item, Exception e, StepContribution contribution) {
if (writeSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
contribution.incrementWriteSkipCount();

80
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTaskletTests.java
View File

@@ -19,6 +19,8 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.easymock.EasyMock.*;
import java.util.ArrayList;
import java.util.List;
@@ -26,6 +28,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.scope.ChunkContext;
@@ -345,9 +348,84 @@ public class FaultTolerantChunkOrientedTaskletTests {
}
catch (Exception e) {
assertEquals(WRITER_FAILED_MESSAGE, e.getMessage());
assertEquals(i*CHUNK_SIZE, processed.size());
assertEquals(i * CHUNK_SIZE, processed.size());
}
}
}
/**
* Make sure skip counts are correct when items are skipped on both process
* and write in the same chunk.
*/
@Test
public void testSkipItemOnProcessAndWrite() throws Exception {
final String WRITER_FAILED_MESSAGE = "writer failed";
final String PROCESSOR_FAILED_MESSAGE = "processor failed";
final RuntimeException writerException = new RuntimeException(WRITER_FAILED_MESSAGE);
final RuntimeException processorException = new RuntimeException(PROCESSOR_FAILED_MESSAGE);
final int CHUNK_SIZE = 2;
tasklet = new FaultTolerantChunkOrientedTasklet<Integer, String>(itemReader,
new ItemProcessor<Integer, String>() {
public String process(Integer item) throws Exception {
if (item == 1) {
throw processorException;
}
processed.add(item);
return String.valueOf(item);
}
}, new ItemWriter<String>() {
public void write(List<? extends String> items) throws Exception {
throw writerException;
}
}, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy);
chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(CHUNK_SIZE));
StepContribution contribution = new StepExecution("foo", null).createStepContribution();
ChunkContext attributes = new ChunkContext();
// mock checks skip listener is called as expected
@SuppressWarnings("unchecked")
SkipListener<Integer, String> skipListener = createStrictMock(SkipListener.class);
tasklet.registerListener(skipListener);
skipListener.onSkipInProcess(1, processorException);
expectLastCall();
skipListener.onSkipInWrite("2", writerException);
expectLastCall();
replay(skipListener);
// processor fails first
try {
tasklet.execute(contribution, attributes);
fail();
}
catch (Exception e) {
assertEquals(PROCESSOR_FAILED_MESSAGE, e.getMessage());
}
// we've only rolled back, nothing has been skipped yet
assertEquals(0, contribution.getProcessSkipCount());
assertEquals(0, contribution.getWriteSkipCount());
try {
tasklet.execute(contribution, attributes);
fail();
}
catch (Exception e) {
assertEquals(WRITER_FAILED_MESSAGE, e.getMessage());
}
// processor skipped failed item, writer fails and causes rollback
assertEquals(1, contribution.getProcessSkipCount());
assertEquals(0, contribution.getWriteSkipCount());
tasklet.execute(contribution, attributes);
// both processor and writer skipped
assertEquals(1, contribution.getProcessSkipCount());
assertEquals(1, contribution.getWriteSkipCount());
verify(skipListener);
}
}