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

IN PROGRESS - issue BATCH-572: Retryable exceptions cannot be skippable

Browse Source 
StatefulRetryStepFactoryBean functionality all moved into SkipLimitStepFactoryBean (the former should be removed). TODO: take some more care with exception types that overlap between retry / skip.
This commit is contained in:
dsyer
2008-05-23 12:36:46 +00:00
parent fd6fad60dc
commit 937e898d24
7 changed files with 297 additions and 329 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java
View File

@@ -21,6 +21,7 @@ import org.springframework.batch.core.StepListener;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.exception.DefaultExceptionHandler;
import org.springframework.batch.repeat.exception.ExceptionHandler;
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.batch.repeat.support.RepeatTemplate;
@@ -57,7 +58,7 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean {
private RepeatTemplate chunkOperations;
private ExceptionHandler exceptionHandler;
private ExceptionHandler exceptionHandler = new DefaultExceptionHandler();
/**

312
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java
View File

@@ -2,14 +2,32 @@ package org.springframework.batch.core.step.item;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.listener.CompositeSkipListener;
import org.springframework.batch.core.step.skip.ItemSkipPolicy;
import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy;
import org.springframework.batch.core.step.skip.SkipLimitExceededException;
import org.springframework.batch.item.ItemKeyGenerator;
import org.springframework.batch.repeat.exception.SimpleLimitExceptionHandler;
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.RetryListener;
import org.springframework.batch.retry.RetryOperations;
import org.springframework.batch.retry.backoff.BackOffPolicy;
import org.springframework.batch.retry.callback.RecoveryRetryCallback;
import org.springframework.batch.retry.policy.ExceptionClassifierRetryPolicy;
import org.springframework.batch.retry.policy.NeverRetryPolicy;
import org.springframework.batch.retry.policy.RecoveryCallbackRetryPolicy;
import org.springframework.batch.retry.policy.SimpleRetryPolicy;
import org.springframework.batch.retry.support.RetryTemplate;
import org.springframework.batch.support.SubclassExceptionClassifier;
/**
* Factory bean for step that provides options for configuring skip behavior.
@@ -43,9 +61,49 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
private ItemKeyGenerator itemKeyGenerator;
// TODO: build this into the retry policy
private int skipCacheCapacity = 1024;
private ItemSkipPolicy itemSkipPolicy;
private int retryLimit;
private Class[] retryableExceptionClasses = new Class[] {};
private BackOffPolicy backOffPolicy;
private RetryListener[] retryListeners;
/**
* Public setter for the retry limit. Each item can be retried up to this
* limit.
* @param retryLimit the retry limit to set
*/
public void setRetryLimit(int retryLimit) {
this.retryLimit = retryLimit;
}
/**
* Public setter for the Class[].
* @param retryableExceptionClasses the retryableExceptionClasses to set
*/
public void setRetryableExceptionClasses(Class[] retryableExceptionClasses) {
this.retryableExceptionClasses = retryableExceptionClasses;
}
/**
* Public setter for the {@link BackOffPolicy}.
* @param backOffPolicy the {@link BackOffPolicy} to set
*/
public void setBackOffPolicy(BackOffPolicy backOffPolicy) {
this.backOffPolicy = backOffPolicy;
}
/**
* Public setter for the {@link RetryListener}s.
* @param retryListeners the {@link RetryListener}s to set
*/
public void setRetryListeners(RetryListener[] retryListeners) {
this.retryListeners = retryListeners;
}
/**
* Public setter for a limit that determines skip policy. If this value is
@@ -80,30 +138,6 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
this.fatalExceptionClasses = fatalExceptionClasses;
}
/**
* Protected getter for the fatal exceptions.
* @return the fatalExceptionClasses
*/
protected Class[] getFatalExceptionClasses() {
return fatalExceptionClasses;
}
/**
* Protected getter for the skippable exceptions.
* @return the skippableExceptionClasses
*/
protected Class[] getSkippableExceptionClasses() {
return skippableExceptionClasses;
}
/**
* Protected getter for the skip limit.
* @return the skipLimit
*/
protected int getSkipLimit() {
return skipLimit;
}
/**
* Public setter for the {@link ItemKeyGenerator}. This is used to identify
* failed items so they can be skipped if encountered again, generally in
@@ -115,22 +149,6 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
this.itemKeyGenerator = itemKeyGenerator;
}
/**
* Protected getter for the {@link ItemKeyGenerator}.
* @return the {@link ItemKeyGenerator}
*/
protected ItemKeyGenerator getItemKeyGenerator() {
return itemKeyGenerator;
}
/**
* Protected getter for the {@link ItemSkipPolicy}.
* @return the itemSkipPolicy
*/
protected ItemSkipPolicy getItemSkipPolicy() {
return itemSkipPolicy;
}
/**
* Public setter for the capacity of the skipped item cache. If a large
* number of items are failing and not being recognized as skipped, it
@@ -168,45 +186,66 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
protected void applyConfiguration(ItemOrientedStep step) {
super.applyConfiguration(step);
ItemSkipPolicyItemHandler itemHandler = new ItemSkipPolicyItemHandler(getItemReader(), getItemWriter());
if (retryLimit > 0 || skipLimit > 0) {
if (skipLimit > 0) {
/*
* If there is a skip limit (not the default) then we are prepared
* to absorb exceptions at the step level because the failed items
* will never re-appear after a rollback.
*/
addFatalExceptionIfMissing(SkipLimitExceededException.class);
List fatalExceptionList = Arrays.asList(fatalExceptionClasses);
addFatalExceptionIfMissing(RetryException.class);
LimitCheckingItemSkipPolicy limitCheckingSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, Arrays
.asList(skippableExceptionClasses), fatalExceptionList);
itemHandler.setItemSkipPolicy(limitCheckingSkipPolicy);
itemHandler.setDoNotRethrowExceptionClasses(noRollbackForExceptionClasses);
this.itemSkipPolicy = limitCheckingSkipPolicy;
SimpleLimitExceptionHandler exceptionHandler = new SimpleLimitExceptionHandler(skipLimit);
exceptionHandler.setExceptionClasses(skippableExceptionClasses);
exceptionHandler.setFatalExceptionClasses(fatalExceptionClasses);
SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy(retryLimit);
if (retryableExceptionClasses.length>0) { // otherwise we retry
// all exceptions
simpleRetryPolicy.setRetryableExceptionClasses(retryableExceptionClasses);
}
simpleRetryPolicy.setFatalExceptionClasses(fatalExceptionClasses);
getStepOperations().setExceptionHandler(exceptionHandler);
ExceptionClassifierRetryPolicy retryPolicy = new ExceptionClassifierRetryPolicy();
SubclassExceptionClassifier exceptionClassifier = new SubclassExceptionClassifier();
HashMap exceptionTypeMap = new HashMap();
for (int i = 0; i < retryableExceptionClasses.length; i++) {
Class cls = retryableExceptionClasses[i];
exceptionTypeMap.put(cls, "retry");
}
exceptionClassifier.setTypeMap(exceptionTypeMap);
HashMap retryPolicyMap = new HashMap();
retryPolicyMap.put("retry", simpleRetryPolicy);
retryPolicyMap.put("default", new NeverRetryPolicy());
retryPolicy.setPolicyMap(retryPolicyMap);
retryPolicy.setExceptionClassifier(exceptionClassifier);
// for subclass to pick up limit and exception classes
setExceptionHandler(exceptionHandler);
// Co-ordinate the retry policy with the exception handler:
getStepOperations().setExceptionHandler(
new SimpleRetryExceptionHandler(retryPolicy, getExceptionHandler(), fatalExceptionClasses));
itemHandler.setItemKeyGenerator(itemKeyGenerator);
itemHandler.setSkipCacheCapacity(skipCacheCapacity);
RecoveryCallbackRetryPolicy recoveryCallbackRetryPolicy = new RecoveryCallbackRetryPolicy(retryPolicy);
recoveryCallbackRetryPolicy.setRecoverableExceptionClasses(noRollbackForExceptionClasses);
BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper();
itemHandler.setSkipListeners(helper.getSkipListeners(getListeners()));
RetryTemplate retryTemplate = new RetryTemplate();
if (retryListeners != null) {
retryTemplate.setListeners(retryListeners);
}
retryTemplate.setRetryPolicy(recoveryCallbackRetryPolicy);
if (backOffPolicy != null) {
retryTemplate.setBackOffPolicy(backOffPolicy);
}
List exceptions = new ArrayList(Arrays.asList(skippableExceptionClasses));
ItemSkipPolicy readSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, exceptions, Arrays
.asList(fatalExceptionClasses));
exceptions.addAll(Arrays.asList(retryableExceptionClasses));
ItemSkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, exceptions, Arrays
.asList(fatalExceptionClasses));
StatefulRetryItemHandler itemHandler = new StatefulRetryItemHandler(getItemReader(), getItemWriter(),
retryTemplate, itemKeyGenerator, readSkipPolicy, writeSkipPolicy);
itemHandler.setSkipListeners(new BatchListenerFactoryHelper().getSkipListeners(getListeners()));
step.setItemHandler(itemHandler);
}
else {
// This is the default in ItemOrientedStep anyway...
itemHandler.setItemSkipPolicy(new NeverSkipItemSkipPolicy());
step.setItemHandler(new SimpleItemHandler(getItemReader(), getItemWriter()));
}
step.setItemHandler(itemHandler);
}
public void addFatalExceptionIfMissing(Class cls) {
@@ -217,4 +256,139 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
fatalExceptionClasses = (Class[]) fatalExceptionList.toArray(new Class[0]);
}
/**
* If there is an exception on input it is skipped if allowed. If there is
* an exception on output, it will be re-thrown in any case, and the
* behaviour when the item is next encountered depends on the retryable and
* skippable exception configuration. If the exception is retryable the
* write will be attempted again up to the retry limit. When retry attempts
* are exhausted the skip listener is invoked and the skip count
* incremented. A retryable exception is thus also effectively also
* implicitly skippable.
*
* @author Dave Syer
*
*/
private static class StatefulRetryItemHandler extends SimpleItemHandler {
final private RetryOperations retryOperations;
final private ItemKeyGenerator itemKeyGenerator;
final private CompositeSkipListener listener = new CompositeSkipListener();
final private ItemSkipPolicy readSkipPolicy;
final private ItemSkipPolicy writeSkipPolicy;
/**
* @param itemReader
* @param itemWriter
* @param retryTemplate
* @param itemKeyGenerator
*/
public StatefulRetryItemHandler(ItemReader itemReader, ItemWriter itemWriter, RetryOperations retryTemplate,
ItemKeyGenerator itemKeyGenerator, ItemSkipPolicy readSkipPolicy, ItemSkipPolicy writeSkipPolicy) {
super(itemReader, itemWriter);
this.retryOperations = retryTemplate;
this.itemKeyGenerator = itemKeyGenerator;
this.readSkipPolicy = readSkipPolicy;
this.writeSkipPolicy = writeSkipPolicy;
}
/**
* Register some {@link SkipListener}s with the handler. Each will get
* the callbacks in the order specified at the correct stage if a skip
* occurs.
*
* @param listeners
*/
public void setSkipListeners(SkipListener[] listeners) {
for (int i = 0; i < listeners.length; i++) {
registerSkipListener(listeners[i]);
}
}
/**
* Register a listener for callbacks at the appropriate stages in a skip
* process.
*
* @param listener a {@link SkipListener}
*/
public void registerSkipListener(SkipListener listener) {
this.listener.register(listener);
}
/**
* Tries to read the item from the reader, in case of exception skip the
* item if the skip policy allows, otherwise re-throw.
*
* @param contribution current StepContribution holding skipped items
* count
* @return next item for processing
*/
protected Object read(StepContribution contribution) throws Exception {
while (true) {
try {
return doRead();
}
catch (Exception e) {
try {
if (readSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
// increment skip count and try again
contribution.incrementTemporaryReadSkipCount();
listener.onSkipInRead(e);
logger.debug("Skipping failed input", e);
}
else {
// re-throw only when the skip policy runs out of
// patience
throw e;
}
}
catch (SkipLimitExceededException ex) {
// we are headed for a abnormal ending so bake in the
// skip count
contribution.combineSkipCounts();
throw ex;
}
}
}
}
/**
* Execute the business logic, delegating to the writer.<br/>
*
* Process the item 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/>
*
* @see org.springframework.batch.core.step.item.SimpleItemHandler#write(java.lang.Object,
* org.springframework.batch.core.StepContribution)
*/
protected void write(final Object item, final StepContribution contribution) throws Exception {
RecoveryRetryCallback retryCallback = new RecoveryRetryCallback(item, new RetryCallback() {
public Object doWithRetry(RetryContext context) throws Throwable {
doWrite(item);
return null;
}
}, itemKeyGenerator != null ? itemKeyGenerator.getKey(item) : item);
retryCallback.setRecoveryCallback(new RecoveryCallback() {
public Object recover(RetryContext context) {
Throwable t = context.getLastThrowable();
if (writeSkipPolicy.shouldSkip(t, contribution.getStepSkipCount())) {
listener.onSkipInWrite(item, t);
}
contribution.incrementWriteSkipCount();
return null;
}
});
retryOperations.execute(retryCallback);
}
}
}

248
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBean.java
View File

@@ -15,32 +15,11 @@
*/
package org.springframework.batch.core.step.item;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.listener.CompositeSkipListener;
import org.springframework.batch.core.step.skip.ItemSkipPolicy;
import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
import org.springframework.batch.core.step.skip.SkipLimitExceededException;
import org.springframework.batch.item.ItemKeyGenerator;
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.RetryListener;
import org.springframework.batch.retry.RetryOperations;
import org.springframework.batch.retry.RetryPolicy;
import org.springframework.batch.retry.backoff.BackOffPolicy;
import org.springframework.batch.retry.callback.RecoveryRetryCallback;
import org.springframework.batch.retry.policy.RecoveryCallbackRetryPolicy;
import org.springframework.batch.retry.policy.SimpleRetryPolicy;
import org.springframework.batch.retry.support.RetryTemplate;
/**
* Factory bean for step that executes its item processing with a stateful
@@ -63,231 +42,4 @@ import org.springframework.batch.retry.support.RetryTemplate;
*/
public class StatefulRetryStepFactoryBean extends SkipLimitStepFactoryBean {
private int retryLimit;
private Class[] retryableExceptionClasses;
private BackOffPolicy backOffPolicy;
private RetryListener[] retryListeners;
/**
* Public setter for the retry limit. Each item can be retried up to this
* limit.
* @param retryLimit the retry limit to set
*/
public void setRetryLimit(int retryLimit) {
this.retryLimit = retryLimit;
}
/**
* Public setter for the Class[].
* @param retryableExceptionClasses the retryableExceptionClasses to set
*/
public void setRetryableExceptionClasses(Class[] retryableExceptionClasses) {
this.retryableExceptionClasses = retryableExceptionClasses;
}
/**
* Public setter for the {@link BackOffPolicy}.
* @param backOffPolicy the {@link BackOffPolicy} to set
*/
public void setBackOffPolicy(BackOffPolicy backOffPolicy) {
this.backOffPolicy = backOffPolicy;
}
/**
* Public setter for the {@link RetryListener}s.
* @param retryListeners the {@link RetryListener}s to set
*/
public void setRetryListeners(RetryListener[] retryListeners) {
this.retryListeners = retryListeners;
}
/**
* @param step
*
*/
protected void applyConfiguration(ItemOrientedStep step) {
super.applyConfiguration(step);
if (retryLimit > 0) {
addFatalExceptionIfMissing(RetryException.class);
SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(retryLimit);
if (retryableExceptionClasses != null) { // otherwise we retry
// all exceptions
retryPolicy.setRetryableExceptionClasses(retryableExceptionClasses);
retryPolicy.setFatalExceptionClasses(getFatalExceptionClasses());
}
// Co-ordinate the retry policy with the exception handler:
getStepOperations().setExceptionHandler(
new SimpleRetryExceptionHandler(retryPolicy, getExceptionHandler(), getFatalExceptionClasses()));
RecoveryCallbackRetryPolicy recoveryCallbackRetryPolicy = new RecoveryCallbackRetryPolicy(retryPolicy);
RetryTemplate retryTemplate = new RetryTemplate();
if (retryListeners != null) {
retryTemplate.setListeners(retryListeners);
}
retryTemplate.setRetryPolicy(recoveryCallbackRetryPolicy);
if (backOffPolicy != null) {
retryTemplate.setBackOffPolicy(backOffPolicy);
}
List exceptions = new ArrayList(Arrays.asList(getSkippableExceptionClasses()));
if (retryableExceptionClasses != null) {
exceptions.addAll(Arrays.asList(retryableExceptionClasses));
}
LimitCheckingItemSkipPolicy itemSkipPolicy = new LimitCheckingItemSkipPolicy(getSkipLimit(), exceptions,
Arrays.asList(getFatalExceptionClasses()));
StatefulRetryItemHandler itemHandler = new StatefulRetryItemHandler(getItemReader(), getItemWriter(),
retryTemplate, getItemKeyGenerator(), itemSkipPolicy);
itemHandler.setSkipListeners(new BatchListenerFactoryHelper().getSkipListeners(getListeners()));
step.setItemHandler(itemHandler);
}
}
/**
* If there is an exception on input it is skipped if allowed. If there is
* an exception on output, it will be re-thrown in any case, and the
* behaviour when the item is next encountered depends on the retryable and
* skippable exception configuration. If the exception is retryable the
* write will be attempted again up to the retry limit. When retry attempts
* are exhausted the skip listener is invoked and the skip count
* incremented. A retryable exception is thus also effectively also
* implicitly skippable.
*
* @author Dave Syer
*
*/
private static class StatefulRetryItemHandler extends SimpleItemHandler {
final private RetryOperations retryOperations;
final private ItemKeyGenerator itemKeyGenerator;
private CompositeSkipListener listener = new CompositeSkipListener();
final private ItemSkipPolicy itemSkipPolicy;
/**
* @param itemReader
* @param itemWriter
* @param retryTemplate
* @param itemKeyGenerator
*/
public StatefulRetryItemHandler(ItemReader itemReader, ItemWriter itemWriter, RetryOperations retryTemplate,
ItemKeyGenerator itemKeyGenerator, ItemSkipPolicy itemSkipPolicy) {
super(itemReader, itemWriter);
this.retryOperations = retryTemplate;
this.itemKeyGenerator = itemKeyGenerator;
this.itemSkipPolicy = itemSkipPolicy;
}
/**
* Register some {@link SkipListener}s with the handler. Each will get
* the callbacks in the order specified at the correct stage if a skip
* occurs.
*
* @param listeners
*/
public void setSkipListeners(SkipListener[] listeners) {
for (int i = 0; i < listeners.length; i++) {
registerSkipListener(listeners[i]);
}
}
/**
* Register a listener for callbacks at the appropriate stages in a skip
* process.
*
* @param listener a {@link SkipListener}
*/
public void registerSkipListener(SkipListener listener) {
this.listener.register(listener);
}
/**
* Tries to read the item from the reader, in case of exception skip the
* item if the skip policy allows, otherwise re-throw.
*
* @param contribution current StepContribution holding skipped items
* count
* @return next item for processing
*/
protected Object read(StepContribution contribution) throws Exception {
while (true) {
try {
return doRead();
}
catch (Exception e) {
try {
if (itemSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
// increment skip count and try again
contribution.incrementTemporaryReadSkipCount();
if (listener != null) {
listener.onSkipInRead(e);
}
logger.debug("Skipping failed input", e);
}
else {
// re-throw only when the skip policy runs out of
// patience
throw e;
}
}
catch (SkipLimitExceededException ex) {
// we are headed for a abnormal ending so bake in the
// skip count
contribution.combineSkipCounts();
throw ex;
}
}
}
}
/**
* Execute the business logic, delegating to the writer.<br/>
*
* Process the item 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/>
*
* @see org.springframework.batch.core.step.item.SimpleItemHandler#write(java.lang.Object,
* org.springframework.batch.core.StepContribution)
*/
protected void write(final Object item, final StepContribution contribution) throws Exception {
RecoveryRetryCallback retryCallback = new RecoveryRetryCallback(item, new RetryCallback() {
public Object doWithRetry(RetryContext context) throws Throwable {
doWrite(item);
return null;
}
}, itemKeyGenerator != null ? itemKeyGenerator.getKey(item) : item);
retryCallback.setRecoveryCallback(new RecoveryCallback() {
public Object recover(RetryContext context) {
Throwable t = context.getLastThrowable();
// TODO: add retryable exceptions as well? (Or ensure that
// all retryable exceptions are skippable?)
if (itemSkipPolicy.shouldSkip(t, contribution.getStepSkipCount())) {
listener.onSkipInWrite(item, t);
}
contribution.incrementWriteSkipCount();
return null;
}
});
retryOperations.execute(retryCallback);
}
}
}