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

RESOLVED - issue BATCH-544: Add a step factory that allows simplified injection of a RetryPolicy

Browse Source
This commit is contained in:
dsyer
2008-07-21 14:07:13 +00:00
parent 705bc284bf
commit c424603e59
4 changed files with 107 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java
View File

@@ -274,20 +274,6 @@ public class ItemOrientedStep extends AbstractStep {
// only if chunk was successful
stepExecution.apply(contribution);
// Attempt to flush before the step execution and stream
// state are updated
try {
itemHandler.flush();
} catch (Error e) {
if (transactionAttribute.rollbackOn(e)) {
throw e;
}
} catch (Exception e) {
if (transactionAttribute.rollbackOn(e)) {
throw e;
}
}
try {
stream.update(stepExecution.getExecutionContext());
} catch (Error e) {
@@ -358,6 +344,7 @@ public class ItemOrientedStep extends AbstractStep {
* @return true if there is more data to process.
*/
protected ExitStatus processChunk(final StepExecution execution, final StepContribution contribution) {
ExitStatus result = chunkOperations.iterate(new RepeatCallback() {
public ExitStatus doInIteration(final RepeatContext context) throws Exception {
if (execution.isTerminateOnly()) {
@@ -371,6 +358,11 @@ public class ItemOrientedStep extends AbstractStep {
return exitStatus;
}
});
// Attempt to flush before the step execution and stream
// state are updated
itemHandler.flush();
return result;
}

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

@@ -41,8 +41,8 @@ import org.springframework.batch.support.SubclassExceptionClassifier;
* to be exclusive.
*
* Skippable exceptions on write will by default cause transaction rollback - to
* avoid rollback for specific exception class include it in the
* transaction attribute as "no rollback for".
* avoid rollback for specific exception class include it in the transaction
* attribute as "no rollback for".
*
* @see SimpleStepFactoryBean
*
@@ -62,7 +62,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
private int cacheCapacity = 0;
private int retryLimit;
private int retryLimit = 0;
private Class<?>[] retryableExceptionClasses = new Class[] {};
@@ -70,6 +70,19 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
private RetryListener[] retryListeners;
private RetryPolicy retryPolicy;
/**
* Setter for the retry policy. If this is specified the other retry
* properties are ignored (retryLimit, backOffPolicy,
* retryableExceptionClasses).
*
* @param retryPolicy a stateless {@link RetryPolicy}
*/
public void setRetryPolicy(RetryPolicy retryPolicy) {
this.retryPolicy = retryPolicy;
}
/**
* Public setter for the retry limit. Each item can be retried up to this
* limit.
@@ -175,31 +188,37 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
protected void applyConfiguration(ItemOrientedStep step) {
super.applyConfiguration(step);
if (retryLimit > 0 || skipLimit > 0) {
if (retryLimit > 0 || skipLimit > 0 || retryPolicy != null) {
addFatalExceptionIfMissing(SkipLimitExceededException.class);
addFatalExceptionIfMissing(RetryException.class);
SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy(retryLimit);
if (retryableExceptionClasses.length > 0) { // otherwise we retry
// all exceptions
simpleRetryPolicy.setRetryableExceptionClasses(retryableExceptionClasses);
}
simpleRetryPolicy.setFatalExceptionClasses(fatalExceptionClasses);
if (retryPolicy == null) {
SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy(retryLimit);
if (retryableExceptionClasses.length > 0) { // otherwise we
// retry
// all exceptions
simpleRetryPolicy.setRetryableExceptionClasses(retryableExceptionClasses);
}
simpleRetryPolicy.setFatalExceptionClasses(fatalExceptionClasses);
ExceptionClassifierRetryPolicy classifierRetryPolicy = new ExceptionClassifierRetryPolicy();
SubclassExceptionClassifier exceptionClassifier = new SubclassExceptionClassifier();
HashMap<Class<?>, String> exceptionTypeMap = new HashMap<Class<?>, String>();
for (int i = 0; i < retryableExceptionClasses.length; i++) {
Class<?> cls = retryableExceptionClasses[i];
exceptionTypeMap.put(cls, "retry");
}
exceptionClassifier.setTypeMap(exceptionTypeMap);
HashMap<String, RetryPolicy> retryPolicyMap = new HashMap<String, RetryPolicy>();
retryPolicyMap.put("retry", simpleRetryPolicy);
retryPolicyMap.put("default", new NeverRetryPolicy());
classifierRetryPolicy.setPolicyMap(retryPolicyMap);
classifierRetryPolicy.setExceptionClassifier(exceptionClassifier);
retryPolicy = classifierRetryPolicy;
ExceptionClassifierRetryPolicy retryPolicy = new ExceptionClassifierRetryPolicy();
SubclassExceptionClassifier exceptionClassifier = new SubclassExceptionClassifier();
HashMap<Class<?>, String> exceptionTypeMap = new HashMap<Class<?>, String>();
for (int i = 0; i < retryableExceptionClasses.length; i++) {
Class<?> cls = retryableExceptionClasses[i];
exceptionTypeMap.put(cls, "retry");
}
exceptionClassifier.setTypeMap(exceptionTypeMap);
HashMap<String, RetryPolicy> retryPolicyMap = new HashMap<String, RetryPolicy>();
retryPolicyMap.put("retry", simpleRetryPolicy);
retryPolicyMap.put("default", new NeverRetryPolicy());
retryPolicy.setPolicyMap(retryPolicyMap);
retryPolicy.setExceptionClassifier(exceptionClassifier);
// Co-ordinate the retry policy with the exception handler:
getStepOperations().setExceptionHandler(
@@ -219,7 +238,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
retryTemplate.setListeners(retryListeners);
}
retryTemplate.setRetryPolicy(recoveryCallbackRetryPolicy);
if (backOffPolicy != null) {
if (retryPolicy == null && backOffPolicy != null) {
retryTemplate.setBackOffPolicy(backOffPolicy);
}

75
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java
View File

@@ -45,6 +45,7 @@ import org.springframework.batch.item.support.AbstractItemWriter;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.batch.retry.RetryException;
import org.springframework.batch.retry.policy.RetryCacheCapacityExceededException;
import org.springframework.batch.retry.policy.SimpleRetryPolicy;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.batch.support.transaction.TransactionAwareProxyFactory;
import org.springframework.transaction.support.TransactionSynchronizationManager;
@@ -70,9 +71,9 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
JobExecution jobExecution;
private ItemWriter processor = new AbstractItemWriter() {
public void write(Object data) throws Exception {
processed.add((String) data);
private ItemWriter<String> processor = new AbstractItemWriter<String>() {
public void write(String data) throws Exception {
processed.add(data);
}
};
@@ -88,7 +89,7 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
factory.setBeanName("step");
factory.setItemReader(new ListItemReader(new ArrayList<String>()));
factory.setItemReader(new ListItemReader<String>(new ArrayList<String>()));
factory.setItemWriter(processor);
factory.setJobRepository(repository);
factory.setTransactionManager(new ResourcelessTransactionManager());
@@ -123,9 +124,9 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
public void testSuccessfulRetryWithReadFailure() throws Exception {
List<String> items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(new String[] { "a", "b", "c" }));
ItemReader provider = new ListItemReader(items) {
public Object read() {
Object item = super.read();
ItemReader<String> provider = new ListItemReader<String>(items) {
public String read() {
String item = super.read();
count++;
if (count == 2) {
throw new RuntimeException("Temporary error - retry for success.");
@@ -154,9 +155,9 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
factory.setSkipLimit(2);
List<String> items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(new String[] { "a", "b", "c", "d", "e", "f" }));
ItemReader provider = new ListItemReader(items) {
public Object read() {
Object item = super.read();
ItemReader<String> provider = new ListItemReader<String>(items) {
public String read() {
String item = super.read();
count++;
if ("b".equals(item) || "d".equals(item)) {
throw new RuntimeException("Read error - planned but skippable.");
@@ -190,9 +191,9 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
factory.setSkipLimit(2);
List<String> items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(new String[] { "a", "b", "c", "d", "e", "f" }));
ItemReader provider = new ListItemReader(items) {
public Object read() {
Object item = super.read();
ItemReader<String> provider = new ListItemReader<String>(items) {
public String read() {
String item = super.read();
logger.debug("Read Called! Item: [" + item + "]");
count++;
return item;
@@ -231,15 +232,53 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
factory.setSkipLimit(0);
List<String> items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(new String[] { "b" }));
ItemReader provider = new ListItemReader(items) {
public Object read() {
Object item = super.read();
ItemReader<String> provider = new ListItemReader<String>(items) {
public String read() {
String item = super.read();
count++;
return item;
}
};
ItemWriter itemWriter = new AbstractItemWriter() {
public void write(Object item) throws Exception {
ItemWriter<String> itemWriter = new AbstractItemWriter<String>() {
public void write(String item) throws Exception {
logger.debug("Write Called! Item: [" + item + "]");
throw new RuntimeException("Write error - planned but retryable.");
}
};
factory.setItemReader(provider);
factory.setItemWriter(itemWriter);
AbstractStep step = (AbstractStep) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
try {
step.execute(stepExecution);
fail("Expected SkipLimitExceededException");
}
catch (SkipLimitExceededException e) {
// expected
}
assertEquals(0, stepExecution.getSkipCount());
// b is processed 4 times plus the null at end
assertEquals(5, count);
assertEquals(0, stepExecution.getItemCount().intValue());
}
@SuppressWarnings("unchecked")
public void testRetryPolicy() throws Exception {
factory.setRetryPolicy(new SimpleRetryPolicy(4));
factory.setSkipLimit(0);
List<String> items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(new String[] { "b" }));
ItemReader<String> provider = new ListItemReader<String>(items) {
public String read() {
String item = super.read();
count++;
return item;
}
};
ItemWriter<String> itemWriter = new AbstractItemWriter<String>() {
public void write(String item) throws Exception {
logger.debug("Write Called! Item: [" + item + "]");
throw new RuntimeException("Write error - planned but retryable.");
}