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

IN PROGRESS - BATCH-888: skip listeners should be called when chunk is about to commit

Browse Source 
moved onSkipInRead(..) calls after write
This commit is contained in:
robokaso
2008-10-27 17:57:24 +00:00
parent b29d264c50
commit 3d8acc66b3
2 changed files with 48 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -52,9 +52,8 @@ import org.springframework.core.AttributeAccessor;
* listener is invoked and the skip count incremented. A retryable exception is
* thus also effectively also implicitly skippable.
*
* Known limitation: ItemProcessor is assumed to be non-transactional. In case
* of rollback caused by error on write the processing phase will not be
* repeated, only the failed write will.
* ItemProcessor is assumed to be transactional. In case of rollback caused by
* error on write the processing phase will be repeated.
*
* @author Dave Syer
* @author Robert Kasanicky
@@ -79,6 +78,8 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
private static final String SKIPPED_INPUTS_KEY = "SKIPPED_INPUTS_BUFFER_KEY";
private static final String SKIPPED_READS_KEY = "SKIPPED_READS_BUFFER_KEY";
public FaultTolerantChunkOrientedTasklet(ItemReader<? extends T> itemReader,
ItemProcessor<? super T, ? extends S> itemProcessor, ItemWriter<? super S> itemWriter,
RepeatOperations chunkOperations, RetryOperations retryTemplate,
@@ -94,7 +95,7 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
}
/**
* Get the next item from {@link #read(StepContribution)} and if not null
* Get the next item from {@link #read(StepContribution, List)} and if not null
* pass the item to {@link #write(List, StepContribution, Map)}. If the
* {@link ItemProcessor} returns null, the write is omitted and another item
* taken from the reader.
@@ -104,18 +105,18 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
*/
public ExitStatus execute(final StepContribution contribution, AttributeAccessor attributes) throws Exception {
// TODO: check flags to see if these need to be saved or not (e.g. JMS
// not)
final List<T> inputs = getBuffer(attributes, INPUT_BUFFER_KEY);
final List<S> outputs = new ArrayList<S>();
ExitStatus result = ExitStatus.CONTINUABLE;
final List<Exception> skippedReads = getBuffer(attributes, SKIPPED_READS_KEY);
if (inputs.isEmpty() && outputs.isEmpty()) {
result = repeatOperations.iterate(new RepeatCallback() {
public ExitStatus doInIteration(final RepeatContext context) throws Exception {
T item = read(contribution);
T item = read(contribution, skippedReads);
if (item == null) {
return ExitStatus.FINISHED;
@@ -131,9 +132,6 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
return result;
}
// store inputs
attributes.setAttribute(INPUT_BUFFER_KEY, inputs);
}
Map<T, Exception> skippedInputs = getSkippedBuffer(attributes, SKIPPED_INPUTS_KEY);
@@ -143,10 +141,17 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
}
Map<S, Exception> skippedOutputs = getSkippedBuffer(attributes, SKIPPED_OUTPUTS_KEY);
// TODO: make sure exceptions get handled by the appropriate handler
outputs.removeAll(skippedOutputs.keySet());
write(outputs, contribution, skippedOutputs);
for (Exception e : skippedReads) {
try {
listener.onSkipInRead(e);
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
}
}
for (Entry<T, Exception> skip : skippedInputs.entrySet()) {
try {
listener.onSkipInProcess(skip.getKey(), skip.getValue());
@@ -156,15 +161,15 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
}
}
for (Entry<S, Exception> entry : skippedOutputs.entrySet()) {
for (Entry<S, Exception> skip : skippedOutputs.entrySet()) {
try {
listener.onSkipInWrite(entry.getKey(), entry.getValue());
listener.onSkipInWrite(skip.getKey(), skip.getValue());
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in skip listener", ex, entry.getValue());
throw new SkipListenerFailedException("Fatal exception in skip listener", ex, skip.getValue());
}
}
// On successful completion clear the attributes to signal that there is
// no more processing
if (outputs.isEmpty()) {
@@ -186,9 +191,10 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
* item if the skip policy allows, otherwise re-throw.
*
* @param contribution current StepContribution holding skipped items count
* @param skippedReads
* @return next item for processing
*/
protected T read(StepContribution contribution) throws Exception {
protected T read(StepContribution contribution, List<Exception> skippedReads) throws Exception {
while (true) {
try {
@@ -199,12 +205,8 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
if (readSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
// increment skip count and try again
contribution.incrementReadSkipCount();
try {
listener.onSkipInRead(e);
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
}
skippedReads.add(e);
logger.debug("Skipping failed input", e);
}
else {
@@ -339,14 +341,11 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
}
/**
* @param attributes
* @param inputBufferKey
* @return
*/
private static <W> List<W> getBuffer(AttributeAccessor attributes, String key) {
if (!attributes.hasAttribute(key)) {
return new ArrayList<W>();
List<W> emptyList = new ArrayList<W>();
attributes.setAttribute(key, emptyList);
return emptyList;
}
@SuppressWarnings("unchecked")
List<W> resource = (List<W>) attributes.getAttribute(key);
@@ -355,9 +354,9 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
private static <E> Map<E, Exception> getSkippedBuffer(AttributeAccessor attributes, String key) {
if (!attributes.hasAttribute(key)) {
Map<E, Exception> result = new LinkedHashMap<E, Exception>();
attributes.setAttribute(key, result);
return result;
Map<E, Exception> emptyMap = new LinkedHashMap<E, Exception>();
attributes.setAttribute(key, emptyMap);
return emptyMap;
}
@SuppressWarnings("unchecked")
Map<E, Exception> resource = (Map<E, Exception>) attributes.getAttribute(key);

32
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java
View File

@@ -54,7 +54,7 @@ public class FaultTolerantStepFactoryBeanTests {
private SkipWriterStub writer = new SkipWriterStub();
private JobExecution jobExecution;
private List<String> processed = new ArrayList<String>();
protected int count;
@@ -80,11 +80,11 @@ public class FaultTolerantStepFactoryBeanTests {
*/
@Test
public void testNonSkippableExceptionOnRead() throws Exception {
// nothing is skippable
Collection<Class<? extends Throwable>> empty = Collections.emptySet();
factory.setSkippableExceptionClasses(empty);
// no exceptions on write
factory.setItemWriter(new ItemWriter<String>() {
public void write(List<? extends String> items) throws Exception {
@@ -97,16 +97,17 @@ public class FaultTolerantStepFactoryBeanTests {
step.execute(stepExecution);
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
//assertEquals("Ouch!", stepExecution.getFailureExceptions().get(0).getMessage());
// assertEquals("Ouch!",
// stepExecution.getFailureExceptions().get(0).getMessage());
}
@Test
public void testNonSkippableException() throws Exception {
// nothing is skippable
Collection<Class<? extends Throwable>> empty = Collections.emptySet();
factory.setSkippableExceptionClasses(empty);
factory.setCommitInterval(1);
// no failures on read
reader = new SkipReaderStub(new String[] { "1", "2", "3", "4", "5" }, new ArrayList<String>());
factory.setItemReader(reader);
@@ -115,7 +116,7 @@ public class FaultTolerantStepFactoryBeanTests {
public void write(List<? extends String> items) throws Exception {
throw new RuntimeException("non-skippable exception");
}
});
Step step = (Step) factory.getObject();
@@ -292,9 +293,11 @@ public class FaultTolerantStepFactoryBeanTests {
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
assertEquals("oops", stepExecution.getFailureExceptions().get(0).getCause().getMessage());
assertEquals(1, stepExecution.getSkipCount());
assertEquals(1, stepExecution.getReadSkipCount());
assertEquals(0, stepExecution.getWriteSkipCount());
// listeners are called only once chunk is about to commit, so
// listener failure does not affect other statistics
assertEquals(3, stepExecution.getSkipCount());
assertEquals(2, stepExecution.getReadSkipCount());
assertEquals(1, stepExecution.getWriteSkipCount());
}
@@ -499,7 +502,7 @@ public class FaultTolerantStepFactoryBeanTests {
assertEquals(2, stepExecution.getSkipCount());
assertEquals(2, stepExecution.getRollbackCount());
}
@Test
public void testReprocessingAfterWriterRollback() throws Exception {
factory.setItemProcessor(new ItemProcessor<String, String>() {
@@ -510,14 +513,15 @@ public class FaultTolerantStepFactoryBeanTests {
});
final Collection<String> NO_FAILURES = Collections.emptyList();
factory.setItemReader(new SkipReaderStub(new String[] { "1", "2", "3", "4" }, NO_FAILURES));
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
step.execute(stepExecution);
//1,2,3,4,3,4,3,4 - two re-processing attempts until the item is identified and skipped
// 1,2,3,4,3,4,3,4 - two re-processing attempts until the item is
// identified and skipped
assertEquals(8, processed.size());
assertEquals("[1, 2, 3, 4, 3, 4, 3, 4]", processed.toString());
}
private static class SkipProcessorStub implements ItemProcessor<String, String> {