From 4716c74c0d059aaa33e2685530c6308623e3772c Mon Sep 17 00:00:00 2001 From: robokaso Date: Fri, 31 Oct 2008 11:22:39 +0000 Subject: [PATCH] IN PROGRESS - BATCH-896: "DRY" FaultTolerantTasklet implementations shared processing and write implementations --- ...ractFaultTolerantChunkOrientedTasklet.java | 168 ++++++++++++++++- .../FaultTolerantChunkOrientedTasklet.java | 171 ++---------------- ...ringFaultTolerantChunkOrientedTasklet.java | 154 +--------------- 3 files changed, 188 insertions(+), 305 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractFaultTolerantChunkOrientedTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractFaultTolerantChunkOrientedTasklet.java index 2f762b9ea..54a978abd 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractFaultTolerantChunkOrientedTasklet.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractFaultTolerantChunkOrientedTasklet.java @@ -1,22 +1,60 @@ package org.springframework.batch.core.step.item; import java.util.ArrayList; +import java.util.Collections; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Map.Entry; +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.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 implementation of the process and write phase of chunk + * processing. + * + * @param input item type + * @param output item type + * + * @see FaultTolerantChunkOrientedTasklet + * @see NonbufferingFaultTolerantChunkOrientedTasklet + * + * @author Robert Kasanicky + */ public abstract class AbstractFaultTolerantChunkOrientedTasklet extends AbstractItemOrientedTasklet { + final private RetryOperations retryOperations; + + final private ItemSkipPolicy writeSkipPolicy; + + final private ItemSkipPolicy processSkipPolicy; + + final private Classifier rollbackClassifier; + public AbstractFaultTolerantChunkOrientedTasklet(ItemReader itemReader, - ItemProcessor itemProcessor, ItemWriter itemWriter) { + ItemProcessor itemProcessor, ItemWriter itemWriter, + RetryOperations retryOperations, ItemSkipPolicy processSkipPolicy, ItemSkipPolicy writeSkipPolicy, + Classifier rollbackClassifier) { + super(itemReader, itemProcessor, itemWriter); + this.retryOperations = retryOperations; + this.processSkipPolicy = processSkipPolicy; + this.writeSkipPolicy = writeSkipPolicy; + this.rollbackClassifier = rollbackClassifier; } /** @@ -92,4 +130,132 @@ public abstract class AbstractFaultTolerantChunkOrientedTasklet extends Ab } return buffer; } + + /** + * Incorporate retry into the item processor stage. + * + * @param inputs the items to process + * @param outputs the items to write + * @param contribution current context + */ + protected void process(final StepContribution contribution, final List inputs, final List outputs, + final Map skippedInputs) throws Exception { + + int filtered = 0; + + for (final I item : inputs) { + + 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.put(item, e); + logger.debug("Skipping after failed process", 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 chunk, final StepContribution contribution, final Map skipped) + throws Exception { + + RetryCallback retryCallback = new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Exception { + doWrite(chunk); + contribution.incrementWriteCount(chunk.size()); + return null; + } + }; + + RecoveryCallback recoveryCallback = new RecoveryCallback() { + + public Object recover(RetryContext context) throws Exception { + if (chunk.size() == 1) { + Exception e = (Exception) context.getLastThrowable(); + O item = chunk.get(0); + checkSkipPolicy(item, e, contribution); + return null; + } + Exception le = (Exception) context.getLastThrowable(); + if (!rollbackClassifier.classify(le)) { + throw new RetryException( + "Invalid retry state during write caused by exception that does not classify for rollback: ", + le); + } + for (O item : chunk) { + try { + doWrite(Collections.singletonList(item)); + contribution.incrementWriteCount(1); + } + catch (Exception e) { + checkSkipPolicy(item, e, contribution); + if (rollbackClassifier.classify(e)) { + throw e; + } + else { + throw new RetryException( + "Invalid retry state during recovery caused by exception that does not classify for rollback: ", + e); + } + } + } + + return null; + + } + + private void checkSkipPolicy(O item, Exception e, StepContribution contribution) { + if (writeSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) { + contribution.incrementWriteSkipCount(); + skipped.put(item, e); + logger.debug("Skipping after failed write", e); + } + else { + throw new RetryException("Non-skippable exception in recoverer", e); + } + } + + }; + + retryOperations.execute(retryCallback, recoveryCallback, new DefaultRetryState(skipped, rollbackClassifier)); + + } } 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 42fb5b83d..27b9aaca5 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 @@ -16,11 +16,9 @@ package org.springframework.batch.core.step.item; import java.util.ArrayList; -import java.util.Collections; import java.util.List; import java.util.Map; -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.NonSkippableReadException; @@ -31,12 +29,7 @@ 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; @@ -61,16 +54,8 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractFaultTolera private final RepeatOperations repeatOperations; - final private RetryOperations retryOperations; - final private ItemSkipPolicy readSkipPolicy; - final private ItemSkipPolicy writeSkipPolicy; - - final private ItemSkipPolicy processSkipPolicy; - - final private Classifier rollbackClassifier; - private static final String SKIPPED_OUTPUTS_KEY = "SKIPPED_OUTPUTS_BUFFER_KEY"; private static final String SKIPPED_INPUTS_KEY = "SKIPPED_INPUTS_BUFFER_KEY"; @@ -82,18 +67,16 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractFaultTolera RepeatOperations chunkOperations, RetryOperations retryTemplate, Classifier rollbackClassifier, ItemSkipPolicy readSkipPolicy, ItemSkipPolicy writeSkipPolicy, ItemSkipPolicy processSkipPolicy) { - super(itemReader, itemProcessor, itemWriter); + + super(itemReader, itemProcessor, itemWriter, retryTemplate, processSkipPolicy, writeSkipPolicy, + rollbackClassifier); this.repeatOperations = chunkOperations; - this.retryOperations = retryTemplate; - this.rollbackClassifier = rollbackClassifier; this.readSkipPolicy = readSkipPolicy; - this.writeSkipPolicy = writeSkipPolicy; - this.processSkipPolicy = processSkipPolicy; } /** - * Get the next item from {@link #read(StepContribution, List)} and if not null - * pass the item to {@link #write(List, StepContribution, Map)}. If the + * 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. * @@ -145,16 +128,14 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractFaultTolera // On successful completion clear the attributes to signal that there is // no more processing - if (outputs.isEmpty()) { - for (String key : attributes.attributeNames()) { - attributes.removeAttribute(key); - } - inputs.clear(); - outputs.clear(); - skippedInputs.clear(); - skippedOutputs.clear(); + for (String key : attributes.attributeNames()) { + attributes.removeAttribute(key); } + inputs.clear(); + skippedInputs.clear(); + skippedOutputs.clear(); + return result; } @@ -191,134 +172,4 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractFaultTolera } - /** - * - * @param inputs the items to process - * @param outputs the items to write - * @param contribution current context - */ - /** - * Incorporate retry into the item processor stage. - */ - protected void process(final StepContribution contribution, final List inputs, final List outputs, - final Map skippedInputs) throws Exception { - - int filtered = 0; - - for (final I item : inputs) { - - RetryCallback retryCallback = new RetryCallback() { - - public S doWithRetry(RetryContext context) throws Exception { - S output = doProcess(item); - return output; - } - - }; - - RecoveryCallback recoveryCallback = new RecoveryCallback() { - - public S recover(RetryContext context) throws Exception { - Exception e = (Exception) context.getLastThrowable(); - if (processSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) { - contribution.incrementProcessSkipCount(); - skippedInputs.put(item, e); - logger.debug("Skipping after failed process", e); - return null; - } - else { - throw new RetryException("Non-skippable exception in recoverer while processing", e); - } - } - - }; - - S 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 chunk, final StepContribution contribution, final Map skipped) - throws Exception { - - RetryCallback retryCallback = new RetryCallback() { - public Object doWithRetry(RetryContext context) throws Exception { - doWrite(chunk); - contribution.incrementWriteCount(chunk.size()); - return null; - } - }; - - RecoveryCallback recoveryCallback = new RecoveryCallback() { - - public Object recover(RetryContext context) throws Exception { - if (chunk.size() == 1) { - Exception e = (Exception) context.getLastThrowable(); - S item = chunk.get(0); - checkSkipPolicy(item, e, contribution); - return null; - } - Exception le = (Exception) context.getLastThrowable(); - if (!rollbackClassifier.classify(le)) { - throw new RetryException( - "Invalid retry state during write caused by exception that does not classify for rollback: ", le); - } - for (S item : chunk) { - try { - doWrite(Collections.singletonList(item)); - contribution.incrementWriteCount(1); - } - catch (Exception e) { - checkSkipPolicy(item, e, contribution); - if (rollbackClassifier.classify(e)) { - throw e; - } - else { - throw new RetryException( - "Invalid retry state during recovery caused by exception 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(); - skipped.put(item, e); - logger.debug("Skipping after failed write", e); - } - else { - throw new RetryException("Non-skippable exception in recoverer", e); - } - } - - }; - - retryOperations.execute(retryCallback, recoveryCallback, new DefaultRetryState(skipped, rollbackClassifier)); - - chunk.clear(); - - } - } 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 index 71f051542..bde7f0f46 100644 --- 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 @@ -1,12 +1,9 @@ package org.springframework.batch.core.step.item; import java.util.ArrayList; -import java.util.Collections; -import java.util.Iterator; import java.util.List; import java.util.Map; -import org.springframework.batch.core.SkipListener; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.step.skip.ItemSkipPolicy; import org.springframework.batch.item.ItemProcessor; @@ -16,12 +13,7 @@ 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; @@ -39,7 +31,8 @@ import org.springframework.core.AttributeAccessor; * @param input item type * @param output item type */ -public class NonbufferingFaultTolerantChunkOrientedTasklet extends AbstractFaultTolerantChunkOrientedTasklet { +public class NonbufferingFaultTolerantChunkOrientedTasklet extends + AbstractFaultTolerantChunkOrientedTasklet { private static final String SKIPPED_INPUTS_KEY = "SKIPPED_INPUTS_KEY"; @@ -49,31 +42,19 @@ public class NonbufferingFaultTolerantChunkOrientedTasklet extends Abstrac private final RepeatOperations repeatOperations; - 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; - } - + super(itemReader, itemProcessor, itemWriter, retryTemplate, processSkipPolicy, writeSkipPolicy, + rollbackClassifier); + this.repeatOperations = chunkOperations; + this.readSkipPolicy = readSkipPolicy; + } /** * Read-process-write a list of items. Uses fault-tolerant read, process and @@ -114,10 +95,10 @@ public class NonbufferingFaultTolerantChunkOrientedTasklet extends Abstrac final Map skippedOutputs = getBufferedSkips(attributes, SKIPPED_OUTPUTS_KEY); outputs.removeAll(skippedOutputs.keySet()); - write(contribution, outputs, skippedOutputs); - + write(outputs, contribution, skippedOutputs); + callSkipListeners(skippedReads, skippedInputs, skippedOutputs); - + return result; } @@ -149,119 +130,4 @@ public class NonbufferingFaultTolerantChunkOrientedTasklet extends Abstrac } - /** - * Incorporate retry and skip into the item processor stage. Any - * {@link SkipListener} provided is called when retry attempts are - * exhausted. Adds failed items into skipped inputs list so that they can be - * filtered if they are encountered again (after rollback). - * - * @param skippedInputs container for items marked for skipping - */ - private void process(final StepContribution contribution, final List inputs, final List outputs, - final Map skippedInputs) 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.put(item, 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); - - } - - /** - * Write the items 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. - * - * Adds failed items into skipped outputs list so that they can be filtered - * if they are encountered again (after rollback). - * - * @param skippedOutputs container for items marked for skipping - */ - private void write(final StepContribution contribution, final List outputs, - final Map skippedOutputs) 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.put(item, 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)); - - } - }