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

IN PROGRESS - BATCH-896: "DRY" FaultTolerantTasklet implementations

Browse Source 
shared processing and write implementations
This commit is contained in:
robokaso
2008-10-31 11:22:39 +00:00
parent 1412135c0b
commit 4716c74c0d
3 changed files with 188 additions and 305 deletions
Download Patch File Download Diff File Expand all files Collapse all files

168
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractFaultTolerantChunkOrientedTasklet.java
View File

@@ -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 <I> input item type
* @param <O> output item type
*
* @see FaultTolerantChunkOrientedTasklet
* @see NonbufferingFaultTolerantChunkOrientedTasklet
*
* @author Robert Kasanicky
*/
public abstract class AbstractFaultTolerantChunkOrientedTasklet<I, O> extends AbstractItemOrientedTasklet<I, O> {
final private RetryOperations retryOperations;
final private ItemSkipPolicy writeSkipPolicy;
final private ItemSkipPolicy processSkipPolicy;
final private Classifier<Throwable, Boolean> rollbackClassifier;
public AbstractFaultTolerantChunkOrientedTasklet(ItemReader<? extends I> itemReader,
ItemProcessor<? super I, ? extends O> itemProcessor, ItemWriter<? super O> itemWriter) {
ItemProcessor<? super I, ? extends O> itemProcessor, ItemWriter<? super O> itemWriter,
RetryOperations retryOperations, ItemSkipPolicy processSkipPolicy, ItemSkipPolicy writeSkipPolicy,
Classifier<Throwable, Boolean> 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<I, O> 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<I> inputs, final List<O> outputs,
final Map<I, Exception> skippedInputs) throws Exception {
int filtered = 0;
for (final I item : inputs) {
RetryCallback<O> retryCallback = new RetryCallback<O>() {
public O doWithRetry(RetryContext context) throws Exception {
O output = doProcess(item);
return output;
}
};
RecoveryCallback<O> recoveryCallback = new RecoveryCallback<O>() {
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.<br/>
*
* 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.<br/>
*/
protected void write(final List<O> chunk, final StepContribution contribution, final Map<O, Exception> skipped)
throws Exception {
RetryCallback<Object> retryCallback = new RetryCallback<Object>() {
public Object doWithRetry(RetryContext context) throws Exception {
doWrite(chunk);
contribution.incrementWriteCount(chunk.size());
return null;
}
};
RecoveryCallback<Object> recoveryCallback = new RecoveryCallback<Object>() {
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));
}
}

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

@@ -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<I, S> 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<Throwable, Boolean> 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<I, S> extends AbstractFaultTolera
RepeatOperations chunkOperations, RetryOperations retryTemplate,
Classifier<Throwable, Boolean> 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<I, S> 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<I, S> 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<I> inputs, final List<S> outputs,
final Map<I, Exception> skippedInputs) throws Exception {
int filtered = 0;
for (final I item : inputs) {
RetryCallback<S> retryCallback = new RetryCallback<S>() {
public S doWithRetry(RetryContext context) throws Exception {
S output = doProcess(item);
return output;
}
};
RecoveryCallback<S> recoveryCallback = new RecoveryCallback<S>() {
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.<br/>
*
* 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.<br/>
*/
protected void write(final List<S> chunk, final StepContribution contribution, final Map<S, Exception> skipped)
throws Exception {
RetryCallback<Object> retryCallback = new RetryCallback<Object>() {
public Object doWithRetry(RetryContext context) throws Exception {
doWrite(chunk);
contribution.incrementWriteCount(chunk.size());
return null;
}
};
RecoveryCallback<Object> recoveryCallback = new RecoveryCallback<Object>() {
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();
}
}

154
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/NonbufferingFaultTolerantChunkOrientedTasklet.java
View File

@@ -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 <I> input item type
* @param <O> output item type
*/
public class NonbufferingFaultTolerantChunkOrientedTasklet<I, O> extends AbstractFaultTolerantChunkOrientedTasklet<I, O> {
public class NonbufferingFaultTolerantChunkOrientedTasklet<I, O> extends
AbstractFaultTolerantChunkOrientedTasklet<I, O> {
private static final String SKIPPED_INPUTS_KEY = "SKIPPED_INPUTS_KEY";
@@ -49,31 +42,19 @@ public class NonbufferingFaultTolerantChunkOrientedTasklet<I, O> 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<Throwable, Boolean> rollbackClassifier;
public NonbufferingFaultTolerantChunkOrientedTasklet(ItemReader<? extends I> itemReader,
ItemProcessor<? super I, ? extends O> itemProcessor, ItemWriter<? super O> itemWriter,
RepeatOperations chunkOperations, RetryOperations retryTemplate,
Classifier<Throwable, Boolean> 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<I, O> extends Abstrac
final Map<O, Exception> 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<I, O> 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<I> inputs, final List<O> outputs,
final Map<I, Exception> skippedInputs) throws Exception {
int filtered = 0;
for (final Iterator<I> iterator = inputs.iterator(); iterator.hasNext();) {
final I item = iterator.next();
RetryCallback<O> retryCallback = new RetryCallback<O>() {
public O doWithRetry(RetryContext context) throws Exception {
O output = doProcess(item);
return output;
}
};
RecoveryCallback<O> recoveryCallback = new RecoveryCallback<O>() {
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<O> outputs,
final Map<O, Exception> skippedOutputs) throws Exception {
RetryCallback<Object> retryCallback = new RetryCallback<Object>() {
public Object doWithRetry(RetryContext context) throws Exception {
doWrite(outputs);
contribution.incrementWriteCount(outputs.size());
return null;
}
};
RecoveryCallback<Object> recoveryCallback = new RecoveryCallback<Object>() {
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));
}
}