diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/NonbufferingFaultTolerantChunkOrientedTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/NonbufferingFaultTolerantChunkOrientedTasklet.java new file mode 100644 index 000000000..b5664a2b7 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/NonbufferingFaultTolerantChunkOrientedTasklet.java @@ -0,0 +1,270 @@ +package org.springframework.batch.core.step.item; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Set; + +import org.springframework.batch.core.SkipListener; +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.step.skip.ItemSkipPolicy; +import org.springframework.batch.core.step.skip.SkipListenerFailedException; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.batch.repeat.RepeatCallback; +import org.springframework.batch.repeat.RepeatContext; +import org.springframework.batch.repeat.RepeatOperations; +import org.springframework.batch.retry.RecoveryCallback; +import org.springframework.batch.retry.RetryCallback; +import org.springframework.batch.retry.RetryContext; +import org.springframework.batch.retry.RetryException; +import org.springframework.batch.retry.RetryOperations; +import org.springframework.batch.retry.support.DefaultRetryState; +import org.springframework.batch.support.Classifier; +import org.springframework.core.AttributeAccessor; + +/** + * Fault-tolerant chunk-oriented tasklet implementation which does not buffer + * items that have been read - the assumption is that item reader is + * transactional and will re-present the items after transaction rollback. + * + * Note that the implementation relies on {@link Object#equals(Object)} + * comparisons for recognizing items on retry/skip. + * + * TODO garbage collection of skipped items + * + * @author Robert Kasanicky + * + * @param input item type + * @param output item type + */ +public class NonbufferingFaultTolerantChunkOrientedTasklet extends AbstractItemOrientedTasklet { + + private final RepeatOperations repeatOperations; + + private final Set skippedInputs = new HashSet(); + + private final Set skippedOutputs = new HashSet(); + + private final RetryOperations retryOperations; + + private final ItemSkipPolicy readSkipPolicy; + + private final ItemSkipPolicy writeSkipPolicy; + + private final ItemSkipPolicy processSkipPolicy; + + private final Classifier rollbackClassifier; + + public NonbufferingFaultTolerantChunkOrientedTasklet(ItemReader itemReader, + ItemProcessor itemProcessor, ItemWriter itemWriter, + RepeatOperations chunkOperations, RetryOperations retryTemplate, + Classifier rollbackClassifier, ItemSkipPolicy readSkipPolicy, + ItemSkipPolicy writeSkipPolicy, ItemSkipPolicy processSkipPolicy) { + super(itemReader, itemProcessor, itemWriter); + this.repeatOperations = chunkOperations; + this.retryOperations = retryTemplate; + this.rollbackClassifier = rollbackClassifier; + this.readSkipPolicy = readSkipPolicy; + this.writeSkipPolicy = writeSkipPolicy; + this.processSkipPolicy = processSkipPolicy; + } + + /** + * Read-process-write a list of items. + */ + public ExitStatus execute(final StepContribution contribution, AttributeAccessor attributes) throws Exception { + ExitStatus result = ExitStatus.CONTINUABLE; + final List inputs = new ArrayList(); + + result = repeatOperations.iterate(new RepeatCallback() { + + public ExitStatus doInIteration(final RepeatContext context) throws Exception { + I item = read(contribution); + + if (item == null) { + return ExitStatus.FINISHED; + } + inputs.add(item); + contribution.incrementReadCount(); + return ExitStatus.CONTINUABLE; + } + }); + + inputs.removeAll(skippedInputs); + + // If there is no input we don't have to do anything more + if (inputs.isEmpty()) { + return result; + } + + List outputs = new ArrayList(); + process(contribution, inputs, outputs); + + outputs.removeAll(skippedOutputs); + + write(outputs, contribution); + + return result; + } + + /** + * Tries to read the item from the reader, in case of exception skip the + * skip listener is called and exception is re-thrown (failed read causes + * rollback automatically because the reader is assumed to be + * transactional). + * + * @param contribution current StepContribution holding skipped items count + * @return next item for processing + */ + protected I read(StepContribution contribution) throws Exception { + + try { + return doRead(); + } + catch (Exception e) { + + 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); + } + logger.debug("Skipping failed input", e); + } + + throw e; + } + + } + + /** + * Incorporate retry into the item processor stage. + * + * @see org.springframework.batch.core.step.item.FaultTolerantChunkOrientedTasklet#process(org.springframework.batch.core.StepContribution, + * org.springframework.batch.core.step.item.Chunk, + * org.springframework.batch.core.step.item.Chunk) + */ + protected void process(final StepContribution contribution, final List inputs, final List outputs) + throws Exception { + + int filtered = 0; + + for (final Iterator iterator = inputs.iterator(); iterator.hasNext();) { + + final I item = iterator.next(); + + RetryCallback retryCallback = new RetryCallback() { + + public O doWithRetry(RetryContext context) throws Exception { + O output = doProcess(item); + return output; + } + + }; + + RecoveryCallback recoveryCallback = new RecoveryCallback() { + + public O recover(RetryContext context) throws Exception { + Exception e = (Exception) context.getLastThrowable(); + if (processSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) { + contribution.incrementProcessSkipCount(); + skippedInputs.add(item); + try { + listener.onSkipInProcess(item, e); + } + catch (RuntimeException ex) { + throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e); + } + return null; + } + else { + throw new RetryException("Non-skippable exception in recoverer while processing", e); + } + } + + }; + + O output = retryOperations.execute(retryCallback, recoveryCallback, new DefaultRetryState(item, + rollbackClassifier)); + if (output != null) { + outputs.add(output); + } + else { + filtered++; + } + + } + + contribution.incrementFilterCount(filtered); + + } + + /** + * Execute the business logic, delegating to the writer.
+ * + * Process the items with the {@link ItemWriter} in a stateful retry. Any + * {@link SkipListener} provided is called when retry attempts are + * exhausted. The listener callback (on write failure) will happen in the + * next transaction automatically.
+ */ + protected void write(final List outputs, final StepContribution contribution) throws Exception { + + RetryCallback retryCallback = new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Exception { + doWrite(outputs); + contribution.incrementWriteCount(outputs.size()); + return null; + } + }; + + RecoveryCallback recoveryCallback = new RecoveryCallback() { + + public Object recover(RetryContext context) throws Exception { + + for (final O item : outputs) { + try { + doWrite(Collections.singletonList(item)); + contribution.incrementWriteCount(1); + } + catch (Exception e) { + if (writeSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) { + contribution.incrementWriteSkipCount(); + skippedOutputs.add(item); + try { + listener.onSkipInWrite(item, e); + } + catch (RuntimeException ex) { + throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e); + } + } + else { + throw new RetryException("Non-skippable exception in recoverer", e); + } + if (rollbackClassifier.classify(e)) { + throw e; + } + else { + logger.error("Exception encountered that does not classify for rollback: ", e); + } + } + } + + return null; + + } + + }; + + retryOperations.execute(retryCallback, recoveryCallback, new DefaultRetryState(outputs, rollbackClassifier)); + + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java index 3f840eb5a..5c8c7eca3 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java @@ -70,6 +70,12 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean private RetryContextCache retryContextCache; + private boolean isReaderTransactional = false; + + public void setIsReaderTransactional(boolean isReaderTransactional) { + this.isReaderTransactional = isReaderTransactional; + } + /** * Setter for the retry policy. If this is specified the other retry * properties are ignored (retryLimit, backOffPolicy, @@ -248,12 +254,23 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean exceptions.addAll(new ArrayList>(retryableExceptionClasses)); ItemSkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, exceptions, new ArrayList>(fatalExceptionClasses)); - FaultTolerantChunkOrientedTasklet tasklet = new FaultTolerantChunkOrientedTasklet(getItemReader(), getItemProcessor(), - getItemWriter(), getChunkOperations(), retryTemplate, rollbackClassifier, readSkipPolicy, - writeSkipPolicy, writeSkipPolicy); - tasklet.setListeners(getListeners()); - step.setTasklet(tasklet); + if (isReaderTransactional) { + NonbufferingFaultTolerantChunkOrientedTasklet tasklet = new NonbufferingFaultTolerantChunkOrientedTasklet( + getItemReader(), getItemProcessor(), getItemWriter(), getChunkOperations(), retryTemplate, + rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); + tasklet.setListeners(getListeners()); + + step.setTasklet(tasklet); + } + else { + FaultTolerantChunkOrientedTasklet tasklet = new FaultTolerantChunkOrientedTasklet( + getItemReader(), getItemProcessor(), getItemWriter(), getChunkOperations(), retryTemplate, + rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); + tasklet.setListeners(getListeners()); + + step.setTasklet(tasklet); + } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanNonBufferingTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanNonBufferingTests.java new file mode 100644 index 000000000..8bb7bf7e6 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanNonBufferingTests.java @@ -0,0 +1,324 @@ +package org.springframework.batch.core.step.item; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; + +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.BatchStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.StepListener; +import org.springframework.batch.core.listener.SkipListenerSupport; +import org.springframework.batch.core.step.JobRepositorySupport; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.support.ListItemReader; +import org.springframework.batch.support.transaction.ResourcelessTransactionManager; +import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; +import org.springframework.util.StringUtils; + +public class SkipLimitStepFactoryBeanNonBufferingTests { + + protected final Log logger = LogFactory.getLog(getClass()); + + private SkipLimitStepFactoryBean factory = new SkipLimitStepFactoryBean(); + + @SuppressWarnings("unchecked") + private Collection> skippableExceptions = new HashSet>(Arrays + .> asList(SkippableException.class, SkippableRuntimeException.class)); + + private List items = Arrays.asList(new String[] { "1", "2", "3", "4", "5" }); + + private ListItemReader reader = new ListItemReader(TransactionAwareProxyFactory + .createTransactionalList(items)); + + private SkipWriterStub writer = new SkipWriterStub(); + + private JobExecution jobExecution; + + int count = 0; + + @Before + public void setUp() throws Exception { + factory.setBeanName("stepName"); + factory.setJobRepository(new JobRepositorySupport()); + factory.setTransactionManager(new ResourcelessTransactionManager()); + factory.setCommitInterval(2); + factory.setItemReader(reader); + factory.setItemWriter(writer); + factory.setSkippableExceptionClasses(skippableExceptions); + factory.setSkipLimit(2); + factory.setIsReaderTransactional(true); + + JobInstance jobInstance = new JobInstance(new Long(1), new JobParameters(), "skipJob"); + jobExecution = new JobExecution(jobInstance); + } + + /** + * Check items causing errors are skipped as expected. + */ + @Test + public void testSkip() throws Exception { + + factory.setSkipLimit(1); + Step step = (Step) factory.getObject(); + + StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + step.execute(stepExecution); + + assertEquals(1, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(1, stepExecution.getWriteSkipCount()); + + // only one exception caused rollback, but more than once because it + // has to go back and split the chunk up to isolate the failed item + assertEquals(2, stepExecution.getRollbackCount()); + + // assertTrue(reader.processed.contains("4")); + assertFalse(writer.written.contains("4")); + + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,2,3,5")); + assertEquals(expectedOutput, writer.written); + + // 5 items + 2 rollbacks re-reading 2 items each time + assertEquals(9, stepExecution.getReadCount()); + + } + + @Test + public void testSkipOverLimit() throws Exception { + SkipProcessorStub processor = new SkipProcessorStub(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("3"))); + processor.rollback = false; + + factory.setItemProcessor(processor); + + factory.setSkipLimit(1); + + Step step = (Step) factory.getObject(); + + StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + + step.execute(stepExecution); + + assertEquals(1, stepExecution.getSkipCount()); + + assertFalse(writer.written.contains("4")); + + // failure on "4" tripped the skip limit so only first chunk was written + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,2")); + assertEquals(expectedOutput, writer.written); + + } + + /** + * Exception in listener causes failure regardless of skip limit. + * @throws Exception + */ + @Test + public void testSkipListenerFailsOnWrite() throws Exception { + + factory.setSkipLimit(7); // some high limit + factory.setItemReader(reader); + factory.setListeners(new StepListener[] { new SkipListenerSupport() { + @Override + public void onSkipInWrite(String item, Throwable t) { + throw new RuntimeException("oops"); + } + } }); + factory.setSkippableExceptionClasses(Collections.> singleton(Exception.class)); + + Step step = (Step) factory.getObject(); + + StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + + step.execute(stepExecution); + assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); + assertEquals("oops", stepExecution.getFailureExceptions().get(0).getCause().getMessage()); + assertEquals(1, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(1, stepExecution.getWriteSkipCount()); + + } + + @Test + public void testSkipOnWriteNotDoubleCounted() throws Exception { + + writer = new SkipWriterStub(Arrays.asList(StringUtils.commaDelimitedListToStringArray("4,5"))); + + factory.setSkipLimit(4); + factory.setItemReader(reader); + factory.setItemWriter(writer); + factory.setCommitInterval(3); // includes all expected skips + + Step step = (Step) factory.getObject(); + + StepExecution stepExecution = jobExecution.createStepExecution(step.getName()); + + step.execute(stepExecution); + assertEquals(2, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(2, stepExecution.getWriteSkipCount()); + + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,2,3")); + assertEquals(expectedOutput, writer.written); + + } + + @Test + public void testDefaultSkipPolicy() throws Exception { + factory.setSkippableExceptionClasses(Collections.> singleton(Exception.class)); + factory.setSkipLimit(1); + List items = Arrays.asList(new String[] { "a", "b", "c" }); + ItemReader provider = new ListItemReader(TransactionAwareProxyFactory.createTransactionalList(items)) { + public String read() { + String item = super.read(); + count++; + if ("b".equals(item)) { + throw new RuntimeException("Read error - planned failure."); + } + return item; + } + }; + factory.setItemReader(provider); + Step step = (Step) factory.getObject(); + + StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + step.execute(stepExecution); + + assertEquals(1, stepExecution.getSkipCount()); + // b is processed once and skipped, plus 1, plus c, plus the null at end + assertEquals(4, count); + } + + /** + * Scenario: Exception in processor that shouldn't cause rollback + */ + @Test + public void testProcessorRollback() throws Exception { + SkipProcessorStub processor = new SkipProcessorStub(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("1,3"))); + factory.setItemProcessor(processor); + + @SuppressWarnings("unchecked") + final Collection NO_FAILURES = Collections.EMPTY_LIST; + factory.setItemWriter(new SkipWriterStub(NO_FAILURES)); + + Step step = (Step) factory.getObject(); + StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + + processor.rollback = false; + step.execute(stepExecution); + assertEquals(2, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getRollbackCount()); + + } + + /** + * Scenario: Exception in processor that should cause rollback + */ + @Test + public void testProcessorNoRollback() throws Exception { + SkipProcessorStub processor = new SkipProcessorStub(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("1,3"))); + factory.setItemProcessor(processor); + + @SuppressWarnings("unchecked") + final Collection NO_FAILURES = Collections.EMPTY_LIST; + factory.setItemWriter(new SkipWriterStub(NO_FAILURES)); + + Step step = (Step) factory.getObject(); + StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + + processor.rollback = true; + step.execute(stepExecution); + assertEquals(2, stepExecution.getSkipCount()); + assertEquals(2, stepExecution.getRollbackCount()); + } + + private static class SkipProcessorStub implements ItemProcessor { + + private final Collection failures; + + private boolean rollback = false; + + public SkipProcessorStub(Collection failures) { + this.failures = failures; + } + + public String process(String item) throws Exception { + if (failures.contains(item)) { + if (rollback) { + throw new SkippableRuntimeException("should cause rollback"); + } + else { + throw new SkippableException("shouldn't cause rollback"); + } + } + return item; + } + + } + + /** + * Simple item writer that supports skip functionality. + */ + private static class SkipWriterStub implements ItemWriter { + + protected final Log logger = LogFactory.getLog(getClass()); + + // simulate transactional output + private List written = TransactionAwareProxyFactory.createTransactionalList(); + + private final Collection failures; + + @SuppressWarnings("unchecked") + public SkipWriterStub() { + this(StringUtils.commaDelimitedListToSet("4")); + } + + /** + * @param failures commaDelimitedListToSet + */ + public SkipWriterStub(Collection failures) { + this.failures = failures; + } + + public void write(List items) throws Exception { + for (String item : items) { + if (failures.contains(item)) { + logger.debug("Throwing write exception on [" + item + "]"); + throw new SkippableRuntimeException("exception in writer"); + } + written.add(item); + } + } + + } + + private static class SkippableException extends Exception { + public SkippableException(String message) { + super(message); + } + } + + private static class SkippableRuntimeException extends RuntimeException { + public SkippableRuntimeException(String message) { + super(message); + } + } + +}