diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java
index 8b11ceee0..47b509f1c 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java
@@ -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();
/**
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 48a89d37a..466ad9a30 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
@@ -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.
+ *
+ * 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.
+ *
+ * @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);
+ }
+
+ }
+
}
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBean.java
index 4b4562155..22691c8ac 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBean.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBean.java
@@ -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.
- *
- * 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.
- *
- * @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);
- }
-
- }
-
}
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java
index f41e43452..094a1b14d 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java
@@ -8,6 +8,8 @@ import java.util.List;
import junit.framework.TestCase;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobParameters;
@@ -34,6 +36,8 @@ import org.springframework.util.StringUtils;
*/
public class SkipLimitStepFactoryBeanTests extends TestCase {
+ protected final Log logger = LogFactory.getLog(getClass());
+
private SkipLimitStepFactoryBean factory = new SkipLimitStepFactoryBean();
private Class[] skippableExceptions = new Class[] { SkippableException.class, SkippableRuntimeException.class };
@@ -73,11 +77,10 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
step.execute(stepExecution);
-
assertEquals(2, stepExecution.getSkipCount());
assertEquals(1, stepExecution.getReadSkipCount().intValue());
assertEquals(1, stepExecution.getWriteSkipCount().intValue());
-
+
// only write exception caused rollback
assertEquals(1, stepExecution.getRollbackCount().intValue());
@@ -192,11 +195,13 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
assertEquals(1, stepExecution.getWriteSkipCount().intValue());
// writer did not skip "2" as it never made it to writer, only "4" did
+ assertFalse(reader.processed.contains("2"));
assertTrue(reader.processed.contains("4"));
- // failure on "4" tripped the skip limit so we never write anything
- // ("1" was written but rolled back)
- List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray(""));
+ // failure on "5" tripped the skip limit but "4" failed on write and was skipped and
+ // RepeatSynchronizationManager.setCompleteOnly() was called in the retry policy to
+ // aggressively commit after a recovery ("1" was written at that point)
+ List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1"));
assertEquals(expectedOutput, writer.written);
}
@@ -289,6 +294,8 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
*/
private static class SkipReaderStub implements ItemReader {
+ protected final Log logger = LogFactory.getLog(getClass());
+
private final String[] items;
private Collection processed = new ArrayList();
@@ -311,22 +318,27 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
public Object read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException {
counter++;
if (counter >= items.length) {
+ logger.debug("Returning null at count=" + counter);
return null;
}
String item = items[counter];
if (failures.contains(item)) {
+ logger.debug("Throwing exception for [" + item + "] at count=" + counter);
throw new SkippableException("exception in reader");
}
processed.add(item);
+ logger.debug("Returning [" + item + "] at count=" + counter);
return item;
}
public void mark() throws MarkFailedException {
+ logger.debug("Marked at count=" + counter);
marked = counter;
}
public void reset() throws ResetFailedException {
counter = marked;
+ logger.debug("Reset at count=" + counter);
}
}
@@ -336,9 +348,11 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
*/
private static class SkipWriterStub implements ItemWriter {
- List written = new ArrayList();
+ protected final Log logger = LogFactory.getLog(getClass());
- int flushIndex = -1;
+ private List written = new ArrayList();
+
+ private int flushIndex = -1;
private final Collection failures;
@@ -355,7 +369,7 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
public void clear() throws ClearFailedException {
for (int i = flushIndex + 1; i < written.size(); i++) {
- written.remove(i);
+ written.remove(written.size()-1);
}
}
@@ -365,6 +379,7 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
public void write(Object item) throws Exception {
if (failures.contains(item)) {
+ logger.debug("Throwing write exception on [" + item + "]");
throw new SkippableRuntimeException("exception in writer");
}
written.add(item);
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 3665b5817..381dcca87 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
@@ -89,6 +89,7 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
factory.setItemWriter(processor);
factory.setJobRepository(repository);
factory.setTransactionManager(new ResourcelessTransactionManager());
+ factory.setRetryableExceptionClasses(new Class[] { Exception.class });
JobSupport job = new JobSupport("jobName");
job.setRestartable(true);
@@ -107,7 +108,15 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
assertTrue(factory.getObject() instanceof Step);
}
- public void testSuccessfulRetry() throws Exception {
+ /**
+ * N.B. this doesn't really test retry, since the retry is only on write
+ * failures, but it does test that read errors are re-presented for another
+ * try when the retryLimit is high enough (it is used to build an exception
+ * handler).
+ *
+ * @throws Exception
+ */
+ public void testSuccessfulRetryWithReadFailure() throws Exception {
List items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(new String[] { "a", "b", "c" }));
ItemReader provider = new ListItemReader(items) {
@@ -122,6 +131,7 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
};
factory.setItemReader(provider);
factory.setRetryLimit(10);
+ factory.setSkippableExceptionClasses(new Class[0]);
AbstractStep step = (AbstractStep) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
@@ -129,7 +139,7 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
assertEquals(0, stepExecution.getSkipCount());
- // b is processed twice, plus 1, plus c, plus the null at end
+ // b is processed twice, plus a, plus c, plus the null at end
assertEquals(5, count);
}
@@ -201,7 +211,8 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
assertEquals(2, recovered.size());
assertEquals(2, stepExecution.getSkipCount());
assertEquals(2, stepExecution.getWriteSkipCount().intValue());
- // each item once, plus 5 failed retries each for b and d, plus the null terminator
+ // each item once, plus 5 failed retries each for b and d, plus the null
+ // terminator
assertEquals(17, count);
}
}
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ExceptionClassifierRetryPolicy.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ExceptionClassifierRetryPolicy.java
index 6ad2282ae..4b2510ea5 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ExceptionClassifierRetryPolicy.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ExceptionClassifierRetryPolicy.java
@@ -135,6 +135,10 @@ public class ExceptionClassifierRetryPolicy extends AbstractStatelessRetryPolicy
}
public boolean canRetry(RetryContext context) {
+ if (this.context==null) {
+ // there was no error yet
+ return true;
+ }
return policy.canRetry(this.context);
}
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ExceptionClassifierRetryPolicyTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ExceptionClassifierRetryPolicyTests.java
index 3ce22f222..67753c6b9 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ExceptionClassifierRetryPolicyTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ExceptionClassifierRetryPolicyTests.java
@@ -52,6 +52,17 @@ public class ExceptionClassifierRetryPolicyTests extends TestCase {
}
}
+ public void testNullContext() throws Exception {
+ Map map = new HashMap();
+ map.put(ExceptionClassifierSupport.DEFAULT, new NeverRetryPolicy());
+ policy.setPolicyMap(map);
+
+ RetryContext context = policy.open(null, null);
+ assertNotNull(context);
+
+ assertTrue(policy.canRetry(context));
+ }
+
public void testClassifierOperates() throws Exception {
Map map = new HashMap();