diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTasklet.java index 818bb2b2e..20f228634 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTasklet.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTasklet.java @@ -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 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 itemReader, ItemProcessor itemProcessor, ItemWriter itemWriter, RepeatOperations chunkOperations, RetryOperations retryTemplate, @@ -94,7 +95,7 @@ public class FaultTolerantChunkOrientedTasklet 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 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 inputs = getBuffer(attributes, INPUT_BUFFER_KEY); final List outputs = new ArrayList(); ExitStatus result = ExitStatus.CONTINUABLE; + final List 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 extends AbstractItemOriente return result; } - // store inputs - attributes.setAttribute(INPUT_BUFFER_KEY, inputs); - } Map skippedInputs = getSkippedBuffer(attributes, SKIPPED_INPUTS_KEY); @@ -143,10 +141,17 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente } Map 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 skip : skippedInputs.entrySet()) { try { listener.onSkipInProcess(skip.getKey(), skip.getValue()); @@ -156,15 +161,15 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente } } - for (Entry entry : skippedOutputs.entrySet()) { + for (Entry 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 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 skippedReads) throws Exception { while (true) { try { @@ -199,12 +205,8 @@ public class FaultTolerantChunkOrientedTasklet 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 extends AbstractItemOriente } - /** - * @param attributes - * @param inputBufferKey - * @return - */ private static List getBuffer(AttributeAccessor attributes, String key) { if (!attributes.hasAttribute(key)) { - return new ArrayList(); + List emptyList = new ArrayList(); + attributes.setAttribute(key, emptyList); + return emptyList; } @SuppressWarnings("unchecked") List resource = (List) attributes.getAttribute(key); @@ -355,9 +354,9 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente private static Map getSkippedBuffer(AttributeAccessor attributes, String key) { if (!attributes.hasAttribute(key)) { - Map result = new LinkedHashMap(); - attributes.setAttribute(key, result); - return result; + Map emptyMap = new LinkedHashMap(); + attributes.setAttribute(key, emptyMap); + return emptyMap; } @SuppressWarnings("unchecked") Map resource = (Map) attributes.getAttribute(key); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java index d8cdeac6a..77b5ac914 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java @@ -54,7 +54,7 @@ public class FaultTolerantStepFactoryBeanTests { private SkipWriterStub writer = new SkipWriterStub(); private JobExecution jobExecution; - + private List processed = new ArrayList(); protected int count; @@ -80,11 +80,11 @@ public class FaultTolerantStepFactoryBeanTests { */ @Test public void testNonSkippableExceptionOnRead() throws Exception { - + // nothing is skippable Collection> empty = Collections.emptySet(); factory.setSkippableExceptionClasses(empty); - + // no exceptions on write factory.setItemWriter(new ItemWriter() { public void write(List 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> 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()); factory.setItemReader(reader); @@ -115,7 +116,7 @@ public class FaultTolerantStepFactoryBeanTests { public void write(List 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() { @@ -510,14 +513,15 @@ public class FaultTolerantStepFactoryBeanTests { }); final Collection 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 {