From c424603e599a9dfe66dc3271ded007ca9665e092 Mon Sep 17 00:00:00 2001 From: dsyer Date: Mon, 21 Jul 2008 14:07:13 +0000 Subject: [PATCH] RESOLVED - issue BATCH-544: Add a step factory that allows simplified injection of a RetryPolicy --- .../core/step/item/ItemOrientedStep.java | 20 ++--- .../step/item/SkipLimitStepFactoryBean.java | 65 ++++++++++------ .../StatefulRetryStepFactoryBeanTests.java | 75 ++++++++++++++----- .../resources/simple-job-launcher-context.xml | 49 +----------- 4 files changed, 107 insertions(+), 102 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java index 0debc2054..5585a76e4 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java @@ -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; } 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 eeb05dc7b..a25ba2841 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 @@ -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, String> exceptionTypeMap = new HashMap, String>(); + 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()); + classifierRetryPolicy.setPolicyMap(retryPolicyMap); + classifierRetryPolicy.setExceptionClassifier(exceptionClassifier); + retryPolicy = classifierRetryPolicy; - ExceptionClassifierRetryPolicy retryPolicy = new ExceptionClassifierRetryPolicy(); - SubclassExceptionClassifier exceptionClassifier = new SubclassExceptionClassifier(); - HashMap, String> exceptionTypeMap = new HashMap, String>(); - 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); // 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); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java index d1b5cf622..148e14ef5 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java @@ -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 processor = new AbstractItemWriter() { + 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())); + factory.setItemReader(new ListItemReader(new ArrayList())); 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 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 provider = new ListItemReader(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 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 provider = new ListItemReader(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 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 provider = new ListItemReader(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 items = TransactionAwareProxyFactory.createTransactionalList(); items.addAll(Arrays.asList(new String[] { "b" })); - ItemReader provider = new ListItemReader(items) { - public Object read() { - Object item = super.read(); + ItemReader provider = new ListItemReader(items) { + public String read() { + String item = super.read(); count++; return item; } }; - ItemWriter itemWriter = new AbstractItemWriter() { - public void write(Object item) throws Exception { + ItemWriter itemWriter = new AbstractItemWriter() { + 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 items = TransactionAwareProxyFactory.createTransactionalList(); + items.addAll(Arrays.asList(new String[] { "b" })); + ItemReader provider = new ListItemReader(items) { + public String read() { + String item = super.read(); + count++; + return item; + } + }; + ItemWriter itemWriter = new AbstractItemWriter() { + public void write(String item) throws Exception { logger.debug("Write Called! Item: [" + item + "]"); throw new RuntimeException("Write error - planned but retryable."); } diff --git a/spring-batch-samples/src/main/resources/simple-job-launcher-context.xml b/spring-batch-samples/src/main/resources/simple-job-launcher-context.xml index f7f077cfc..fc4681264 100644 --- a/spring-batch-samples/src/main/resources/simple-job-launcher-context.xml +++ b/spring-batch-samples/src/main/resources/simple-job-launcher-context.xml @@ -7,7 +7,8 @@ http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.5.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd"> - + + @@ -23,52 +24,6 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -