From 36801d7d5df51c3761d394479b8a3d8b6601cc87 Mon Sep 17 00:00:00 2001 From: dsyer Date: Mon, 3 Mar 2008 17:50:16 +0000 Subject: [PATCH] OPEN - issue BATCH-404: FactoryBeans for step configuration http://jira.springframework.org/browse/BATCH-404 SPlit item processor implementations --- .../step/support/DefaultStepFactoryBean.java | 19 ++++- ....java => ItemSkipPolicyItemProcessor.java} | 56 ++------------- .../support/StatefulRetryStepFactoryBean.java | 71 ++++++++++++++++--- .../src/main/resources/jobs/retrySample.xml | 5 +- 4 files changed, 83 insertions(+), 68 deletions(-) rename spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/{KitchenSinkItemProcessor.java => ItemSkipPolicyItemProcessor.java} (51%) diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepFactoryBean.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepFactoryBean.java index 2c5beda24..e1bc95040 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepFactoryBean.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepFactoryBean.java @@ -52,6 +52,8 @@ public class DefaultStepFactoryBean extends AbstractStepFactoryBean { private ItemProcessor itemProcessor; + private RepeatTemplate stepOperations; + /** * Set the commit interval. * @@ -96,6 +98,15 @@ public class DefaultStepFactoryBean extends AbstractStepFactoryBean { this.listeners = listeners; } + /** + * Protected getter for the step operations to make them available in + * subclasses. + * @return the step operations + */ + protected RepeatTemplate getStepOperations() { + return stepOperations; + } + /** * Public setter for the {@link TaskExecutor}. If this is set, then it will * be used to execute the chunk processing inside the {@link Step}. @@ -172,7 +183,7 @@ public class DefaultStepFactoryBean extends AbstractStepFactoryBean { StepListener[] stepListeners = helper.getStepListeners(listeners); itemReader = helper.getItemReader(itemReader, listeners); itemWriter = helper.getItemWriter(itemWriter, listeners); - RepeatTemplate stepOperations = new RepeatTemplate(); + stepOperations = new RepeatTemplate(); stepOperations = (RepeatTemplate) helper.getStepOperations(stepOperations, listeners); // In case they are used by subclasses: @@ -185,10 +196,11 @@ public class DefaultStepFactoryBean extends AbstractStepFactoryBean { TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate(); repeatTemplate.setTaskExecutor(taskExecutor); stepOperations = repeatTemplate; - step.setStepOperations(stepOperations); } - KitchenSinkItemProcessor itemProcessor = new KitchenSinkItemProcessor(itemReader, itemWriter); + step.setStepOperations(stepOperations); + + ItemSkipPolicyItemProcessor itemProcessor = new ItemSkipPolicyItemProcessor(itemReader, itemWriter); if (skipLimit > 0) { /* @@ -209,4 +221,5 @@ public class DefaultStepFactoryBean extends AbstractStepFactoryBean { step.setItemProcessor(itemProcessor); } + } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/KitchenSinkItemProcessor.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemSkipPolicyItemProcessor.java similarity index 51% rename from spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/KitchenSinkItemProcessor.java rename to spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemSkipPolicyItemProcessor.java index 8b5f8f772..339b22373 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/KitchenSinkItemProcessor.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemSkipPolicyItemProcessor.java @@ -19,23 +19,14 @@ import org.springframework.batch.core.domain.ItemSkipPolicy; import org.springframework.batch.core.domain.StepContribution; import org.springframework.batch.io.Skippable; import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemRecoverer; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.repeat.ExitStatus; -import org.springframework.batch.retry.RetryCallback; -import org.springframework.batch.retry.RetryOperations; -import org.springframework.batch.retry.callback.ItemReaderRetryCallback; -import org.springframework.batch.retry.support.RetryTemplate; /** * @author Dave Syer * */ -public class KitchenSinkItemProcessor extends SimpleItemProcessor { - - private RetryOperations retryOperations = new RetryTemplate(); - - private ItemReaderRetryCallback retryCallback; +public class ItemSkipPolicyItemProcessor extends SimpleItemProcessor { private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy(); @@ -43,7 +34,7 @@ public class KitchenSinkItemProcessor extends SimpleItemProcessor { * @param itemReader * @param itemWriter */ - public KitchenSinkItemProcessor(ItemReader itemReader, ItemWriter itemWriter) { + public ItemSkipPolicyItemProcessor(ItemReader itemReader, ItemWriter itemWriter) { super(itemReader, itemWriter); } @@ -54,37 +45,13 @@ public class KitchenSinkItemProcessor extends SimpleItemProcessor { this.itemSkipPolicy = itemSkipPolicy; } - /** - * Public setter for the {@link RetryOperations}. - * @param retryOperations the {@link RetryOperations} to set - */ - public void setRetryOperations(RetryOperations retryOperations) { - this.retryOperations = retryOperations; - } - - /** - * Public setter for the ItemReaderRetryCallback. TODO: get rid of this. - * @param retryCallback the retryCallback to set - */ - public void setRetryCallback(ItemReaderRetryCallback retryCallback) { - this.retryCallback = retryCallback; - } - /** * Execute the business logic, delegating to the reader and writer. * Subclasses could extend the behaviour as long as they always return the * value of this method call in their superclass.
* * Read from the {@link ItemReader} and process (if not null) with the - * {@link ItemWriter}. If a {@link RetryCallback} is provided, then the - * call to {@link ItemWriter} is wrapped in a stateful retry. In that case - * the {@link ItemRecoverer} is used (if provided) in the case of an - * exception to apply alternate processing to the item. If the stateful - * retry is in place then the recovery will happen in the next transaction - * automatically, otherwise it might be necessary for clients to make the - * recover method transactional with appropriate propagation behaviour - * (probably REQUIRES_NEW because the call will happen in the context of a - * transaction that is about to rollback).
+ * {@link ItemWriter}.
* * If there is an exception and the reader or writer implements * {@link Skippable} then the skip method is called. @@ -96,10 +63,6 @@ public class KitchenSinkItemProcessor extends SimpleItemProcessor { public ExitStatus process(StepContribution contribution) throws Exception { ExitStatus exitStatus = ExitStatus.CONTINUABLE; - if (retryCallback != null) { - return new ExitStatus(retryOperations.execute(retryCallback) != null); - } - try { exitStatus = super.process(contribution); @@ -107,7 +70,7 @@ public class KitchenSinkItemProcessor extends SimpleItemProcessor { } catch (Exception e) { - if (retryCallback == null && itemSkipPolicy.shouldSkip(e, contribution.getSkipCount())) { + if (itemSkipPolicy.shouldSkip(e, contribution.getSkipCount())) { contribution.incrementSkipCount(); skip(); } @@ -122,19 +85,12 @@ public class KitchenSinkItemProcessor extends SimpleItemProcessor { } /** - * Mark the current item as skipped if possible. If there is a retry policy - * in action there is no need to take any action now because it will be - * covered by the retry in the next transaction. Otherwise if the reader and / - * or writer are {@link Skippable} then delegate to them in that order. + * Mark the current item as skipped if possible. If the reader and / or + * writer are {@link Skippable} then delegate to them in that order. * * @see org.springframework.batch.io.Skippable#skip() */ private void skip() { - if (retryCallback != null) { - // No need to skip because the recoverer will take any action - // necessary. - return; - } if (getItemReader() instanceof Skippable) { ((Skippable) getItemReader()).skip(); } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StatefulRetryStepFactoryBean.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StatefulRetryStepFactoryBean.java index 79ebf18ec..5d5d8bddb 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StatefulRetryStepFactoryBean.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StatefulRetryStepFactoryBean.java @@ -16,10 +16,15 @@ package org.springframework.batch.execution.step.support; import org.springframework.batch.core.domain.Step; +import org.springframework.batch.core.domain.StepContribution; import org.springframework.batch.execution.step.ItemOrientedStep; import org.springframework.batch.item.ItemKeyGenerator; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemRecoverer; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHandler; +import org.springframework.batch.retry.RetryOperations; import org.springframework.batch.retry.RetryPolicy; import org.springframework.batch.retry.callback.ItemReaderRetryCallback; import org.springframework.batch.retry.policy.ItemReaderRetryPolicy; @@ -74,25 +79,25 @@ public class StatefulRetryStepFactoryBean extends DefaultStepFactoryBean { */ protected void applyConfiguration(ItemOrientedStep step) { - // Ensure exception handler always rethrows. N.B. no skips ever actually - // take place. - if (retryPolicy != null) { - // TODO: actually we need to co-ordinate the retry policy with the - // exception handler limit, so this is a hack for now. - super.setSkipLimit(Integer.MAX_VALUE); - } - super.applyConfiguration(step); if (retryPolicy != null) { + + // TODO: actually we need to co-ordinate the retry policy with the + // exception handler limit, so this is a hack for now. + getStepOperations().setExceptionHandler(new SimpleLimitExceptionHandler(Integer.MAX_VALUE)); + ItemReaderRetryCallback retryCallback = new ItemReaderRetryCallback(getItemReader(), getKeyGenerator(), getItemWriter()); ItemReaderRetryPolicy itemProviderRetryPolicy = new ItemReaderRetryPolicy(retryPolicy); + RetryTemplate retryTemplate = new RetryTemplate(); retryTemplate.setRetryPolicy(itemProviderRetryPolicy); - KitchenSinkItemProcessor itemProcessor = (KitchenSinkItemProcessor) getItemProcessor(); - itemProcessor.setRetryOperations(retryTemplate); - itemProcessor.setRetryCallback(retryCallback); + + StatefulRetryItemProcessor itemProcessor = new StatefulRetryItemProcessor(getItemReader(), getItemWriter(), retryTemplate, retryCallback); + + step.setItemProcessor(itemProcessor); + } } @@ -115,4 +120,48 @@ public class StatefulRetryStepFactoryBean extends DefaultStepFactoryBean { } + private static class StatefulRetryItemProcessor extends SimpleItemProcessor { + + final private RetryOperations retryOperations; + + final private ItemReaderRetryCallback retryCallback; + + /** + * @param itemReader + * @param itemWriter + * @param retryCallback + * @param retryTemplate + */ + public StatefulRetryItemProcessor(ItemReader itemReader, ItemWriter itemWriter, RetryOperations retryTemplate, ItemReaderRetryCallback retryCallback) { + super(itemReader, itemWriter); + this.retryOperations = retryTemplate; + this.retryCallback = retryCallback; + } + + /** + * Execute the business logic, delegating to the reader and writer. + * Subclasses could extend the behaviour as long as they always return + * the value of this method call in their superclass.
+ * + * Read from the {@link ItemReader} and process (if not null) with the + * {@link ItemWriter}. The call to {@link ItemWriter} is wrapped in a + * stateful retry. In that case the {@link ItemRecoverer} is used (if + * provided) in the case of an exception to apply alternate processing + * to the item. If the stateful retry is in place then the recovery will + * happen in the next transaction automatically, otherwise it might be + * necessary for clients to make the recover method transactional with + * appropriate propagation behaviour (probably REQUIRES_NEW because the + * call will happen in the context of a transaction that is about to + * rollback).
+ * + * @param contribution the current step + * @return {@link ExitStatus#CONTINUABLE} if there is more processing to + * do + * @throws Exception if there is an error + */ + public ExitStatus process(StepContribution contribution) throws Exception { + return new ExitStatus(retryOperations.execute(retryCallback) != null); + } + + } } diff --git a/spring-batch-samples/src/main/resources/jobs/retrySample.xml b/spring-batch-samples/src/main/resources/jobs/retrySample.xml index 19703d7c0..b03ee2426 100644 --- a/spring-batch-samples/src/main/resources/jobs/retrySample.xml +++ b/spring-batch-samples/src/main/resources/jobs/retrySample.xml @@ -27,14 +27,11 @@ - - -