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 @@
-
-
-