From e5cccef3d12798e155334aa12a2175ee558f742c Mon Sep 17 00:00:00 2001 From: dsyer Date: Sun, 2 Mar 2008 13:07:04 +0000 Subject: [PATCH] OPEN - issue BATCH-404: FactoryBeans for step configuration http://jira.springframework.org/browse/BATCH-404 Remove references to KitchenSinkStepFactoryBean - use StatefulRetryStepFactoryBean for remaining sample. --- .../batch/core/domain/ItemSkipPolicy.java | 13 +- .../execution/step/ItemOrientedStep.java | 14 +- .../support/AlwaysSkipItemSkipPolicy.java | 5 +- .../step/support/DefaultStepFactoryBean.java | 64 ++++++- .../support/KitchenSinkStepFactoryBean.java | 172 ------------------ .../support/LimitCheckingItemSkipPolicy.java | 70 +++---- .../step/support/NeverSkipItemSkipPolicy.java | 5 +- .../step/support/SimpleStepFactoryBean.java | 29 --- .../support/SkipLimitExceededException.java | 4 +- .../support/StatefulRetryStepFactoryBean.java | 117 ++++++++++++ .../support/SimpleStepFactoryBeanTests.java | 2 +- .../src/main/resources/jobs/parallelJob.xml | 2 +- .../src/main/resources/jobs/retrySample.xml | 9 +- 13 files changed, 223 insertions(+), 283 deletions(-) delete mode 100644 spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/KitchenSinkStepFactoryBean.java create mode 100644 spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StatefulRetryStepFactoryBean.java diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/ItemSkipPolicy.java b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/ItemSkipPolicy.java index e3dc7328e..97e22939f 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/ItemSkipPolicy.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/ItemSkipPolicy.java @@ -27,20 +27,11 @@ public interface ItemSkipPolicy { * Returns true or false, indicating whether or not reading should * continue for the current step execution with the given throwable. * - * @param exception throwable encountered while reading + * @param t throwable encountered while reading * @param skipCount currently running count of skips * @return true if reading should continue, false otherwise. * @throws IllegalArgumentException if the exception is null */ - boolean shouldSkip(Exception exception, int skipCount); + boolean shouldSkip(Throwable t, int skipCount); - /** - * Returns true or false, indicating whether or not an exception - * should cause the step to fail. - * - * @param t throwable encountered while reading - * @return true if the step should continue processing, false otherwise - * @throws IllegalArgumentException if the exception is null - */ - boolean shouldFail(Throwable t); } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java index 8234c570a..e90abc3fa 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java @@ -293,6 +293,7 @@ public class ItemOrientedStep extends AbstractStep { .getTransaction(new DefaultTransactionDefinition()); try { + itemReader.mark(); listener.beforeChunk(); result = processChunk(contribution); @@ -364,16 +365,11 @@ public class ItemOrientedStep extends AbstractStep { stepExecution.setStatus(BatchStatus.UNKNOWN); } - if (itemSkipPolicy.shouldFail(t)) { - if (t instanceof RuntimeException) { - throw (RuntimeException) t; - } - else { - throw new RuntimeException(t); - } + if (t instanceof RuntimeException) { + throw (RuntimeException) t; } else { - logger.error("Exception should not cause step to fail", t); + throw new RuntimeException(t); } } @@ -517,7 +513,7 @@ public class ItemOrientedStep extends AbstractStep { } catch (Exception e) { - if (itemSkipPolicy.shouldSkip(e, contribution.getSkipCount())) { + if (retryCallback == null && itemSkipPolicy.shouldSkip(e, contribution.getSkipCount())) { contribution.incrementSkipCount(); skip(); } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/AlwaysSkipItemSkipPolicy.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/AlwaysSkipItemSkipPolicy.java index 1d4ff0797..72819a8b3 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/AlwaysSkipItemSkipPolicy.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/AlwaysSkipItemSkipPolicy.java @@ -26,11 +26,8 @@ import org.springframework.batch.core.domain.ItemSkipPolicy; */ public class AlwaysSkipItemSkipPolicy implements ItemSkipPolicy { - public boolean shouldSkip(Exception ex, int skipCount) { + public boolean shouldSkip(Throwable t, int skipCount) { return true; } - public boolean shouldFail(Throwable t) { - return true; - } } 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 b42c330f9..fd4f73c92 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 @@ -15,31 +15,91 @@ */ package org.springframework.batch.execution.step.support; +import org.springframework.batch.core.domain.BatchListener; +import org.springframework.batch.core.domain.Step; import org.springframework.batch.execution.step.ItemOrientedStep; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHandler; +import org.springframework.batch.repeat.support.RepeatTemplate; +import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate; +import org.springframework.core.task.TaskExecutor; /** * Adds listeners to {@link SimpleStepFactoryBean}. * * @author Dave Syer - * + * */ public class DefaultStepFactoryBean extends SimpleStepFactoryBean { + private boolean alwaysSkip = false; + private Object[] listeners = new Object[0]; + private TaskExecutor taskExecutor; + /** - * @param listeners + * Public setter for the flag that determines skip policy. If this flag is + * true then an exception in chunk processing will cause the item to be + * skipped and no exceptions propagated. If it is false then all exceptions + * will be propagated from the chunk and cause the step to abort. + * + * @param alwaysSkip the value to set. Default is false. + */ + public void setAlwaysSkip(boolean alwaysSkip) { + this.alwaysSkip = alwaysSkip; + } + + /** + * The listeners to inject into the {@link Step}. Any instance of + * {@link BatchListener} or {@link ItemStream} can be used, and will then + * receive callbacks at the appropriate stage in the step. + * + * @param listeners an array of listeners */ public void setListeners(Object[] listeners) { this.listeners = listeners; } + /** + * Public setter for the {@link TaskExecutor}. If this is set, then it will + * be used to execute the chunk processing inside the {@link Step}. + * + * @param taskExecutor the taskExecutor to set + */ + public void setTaskExecutor(TaskExecutor taskExecutor) { + this.taskExecutor = taskExecutor; + } + /** * @param step * */ protected void applyConfiguration(ItemOrientedStep step) { + super.applyConfiguration(step); step.setListeners(listeners); + + RepeatTemplate stepOperations = new RepeatTemplate(); + + if (taskExecutor != null) { + TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate(); + repeatTemplate.setTaskExecutor(taskExecutor); + stepOperations = repeatTemplate; + } + + if (alwaysSkip) { + // If we always skip (not the default) then we are prepared to + // absorb all exceptions at the step level because the failed items + // will never re-appear after a rollback. + step.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy()); + stepOperations.setExceptionHandler(new SimpleLimitExceptionHandler(Integer.MAX_VALUE)); + step.setStepOperations(stepOperations); + } + else { + // This is the default in ItemOrientedStep anyway... + step.setItemSkipPolicy(new NeverSkipItemSkipPolicy()); + } + } } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/KitchenSinkStepFactoryBean.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/KitchenSinkStepFactoryBean.java deleted file mode 100644 index a849d2b2e..000000000 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/KitchenSinkStepFactoryBean.java +++ /dev/null @@ -1,172 +0,0 @@ -/* - * Copyright 2006-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.springframework.batch.execution.step.support; - -import org.springframework.batch.core.domain.ItemSkipPolicy; -import org.springframework.batch.core.domain.Step; -import org.springframework.batch.execution.step.ItemOrientedStep; -import org.springframework.batch.item.ItemKeyGenerator; -import org.springframework.batch.repeat.exception.handler.ExceptionHandler; -import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; -import org.springframework.batch.repeat.support.RepeatTemplate; -import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate; -import org.springframework.batch.retry.RetryPolicy; -import org.springframework.batch.retry.callback.ItemReaderRetryCallback; -import org.springframework.batch.retry.policy.ItemReaderRetryPolicy; -import org.springframework.batch.retry.support.RetryTemplate; -import org.springframework.core.task.TaskExecutor; - -/** - * @author Dave Syer - * - */ -public class KitchenSinkStepFactoryBean extends AbstractStepFactoryBean { - - private int commitInterval = 0; - - private Object[] listeners = new Object[0]; - - private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy(); - - private TaskExecutor taskExecutor; - - private RetryPolicy retryPolicy; - - private ExceptionHandler exceptionHandler; - - private ItemKeyGenerator itemKeyGenerator; - - /** - * Set the commit interval. - * - * @param commitInterval - */ - public void setCommitInterval(int commitInterval) { - this.commitInterval = commitInterval; - } - - /** - * @param listeners - */ - public void setListeners(Object[] listeners) { - this.listeners = listeners; - } - - /** - * @param itemSkipPolicy - */ - public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) { - this.itemSkipPolicy = itemSkipPolicy; - } - - /** - * Public setter for the {@link TaskExecutor}. If this is set, then it will - * be used to execute the chunk processing inside the {@link Step}. - * - * @param taskExecutor the taskExecutor to set - */ - public void setTaskExecutor(TaskExecutor taskExecutor) { - this.taskExecutor = taskExecutor; - } - - /** - * Public setter for the {@link RetryPolicy}. - * @param retryPolicy the {@link RetryPolicy} to set - */ - public void setRetryPolicy(RetryPolicy retryPolicy) { - this.retryPolicy = retryPolicy; - } - - /** - * Set the {@link ExceptionHandler} for the step operations (outer loop). - * @param exceptionHandler - */ - public void setExceptionHandler(ExceptionHandler exceptionHandler) { - this.exceptionHandler = exceptionHandler; - } - - /** - * Public setter for the {@link ItemKeyGenerator}. If it is not injected - * but the reader or writer implement {@link ItemKeyGenerator}, one of - * those will be used instead (preferring the reader to the writer if both - * would be appropriate). - * - * @param itemKeyGenerator the {@link ItemKeyGenerator} to set - */ - public void setItemKeyGenerator(ItemKeyGenerator itemKeyGenerator) { - this.itemKeyGenerator = itemKeyGenerator; - } - - /** - * @param step - * - */ - protected void applyConfiguration(ItemOrientedStep step) { - - super.applyConfiguration(step); - - step.setListeners(listeners); - step.setItemSkipPolicy(itemSkipPolicy); - - if (retryPolicy != null) { - ItemReaderRetryCallback retryCallback = new ItemReaderRetryCallback(getItemReader(), getKeyGenerator(), - getItemWriter()); - ItemReaderRetryPolicy itemProviderRetryPolicy = new ItemReaderRetryPolicy(retryPolicy); - RetryTemplate template = new RetryTemplate(); - template.setRetryPolicy(itemProviderRetryPolicy); - step.setRetryOperations(template); - step.setRetryCallback(retryCallback); - } - - if (commitInterval > 0) { - RepeatTemplate chunkOperations = new RepeatTemplate(); - chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(commitInterval)); - step.setChunkOperations(chunkOperations); - } - - RepeatTemplate stepOperations = new RepeatTemplate(); - - if (taskExecutor != null) { - TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate(); - repeatTemplate.setTaskExecutor(taskExecutor); - stepOperations = repeatTemplate; - } - - if (exceptionHandler != null) { - ((RepeatTemplate) stepOperations).setExceptionHandler(exceptionHandler); - step.setStepOperations(stepOperations); - } - } - - /** - * @return an {@link ItemKeyGenerator} or null if none is found. - */ - private ItemKeyGenerator getKeyGenerator() { - - if (itemKeyGenerator != null) { - return itemKeyGenerator; - } - if (getItemReader() instanceof ItemKeyGenerator) { - return (ItemKeyGenerator) getItemReader(); - } - if (getItemWriter() instanceof ItemKeyGenerator) { - return (ItemKeyGenerator) getItemWriter(); - } - return null; - - } - -} diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/LimitCheckingItemSkipPolicy.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/LimitCheckingItemSkipPolicy.java index b39ecb554..9c63179bc 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/LimitCheckingItemSkipPolicy.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/LimitCheckingItemSkipPolicy.java @@ -30,19 +30,25 @@ import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.io.exception.FlatFileParsingException; /** - *

{@link ItemSkipPolicy} that determines whether or not reading should - * continue based upon how many items have been skipped. This is extremely - * useful behavior, as it allows you to skip records, but will throw a - * {@link SkipLimitExceededException} if a set limit has been exceeded. For example, - * it is generally advisable to skip {@link FlatFileParsingException}s, however, if - * the vast majority of records are causing exceptions, the file is likely bad.

+ *

+ * {@link ItemSkipPolicy} that determines whether or not reading should continue + * based upon how many items have been skipped. This is extremely useful + * behavior, as it allows you to skip records, but will throw a + * {@link SkipLimitExceededException} if a set limit has been exceeded. For + * example, it is generally advisable to skip {@link FlatFileParsingException}s, + * however, if the vast majority of records are causing exceptions, the file is + * likely bad. + *

* - *

Furthermore, it is also likely that you only want to skip certain exceptions. - * {@link FlatFileParsingException} is a good example of an exception you will likely - * want to skip, but a {@link FileNotFoundException} should cause immediate termination - * of the {@link Step}. Because it would be impossible for a general purpose policy to - * determine all the types of exceptions that should be skipped from those that shouldn't, - * a list must be passed in, with all of the exceptions that are 'skippable'

+ *

+ * Furthermore, it is also likely that you only want to skip certain exceptions. + * {@link FlatFileParsingException} is a good example of an exception you will + * likely want to skip, but a {@link FileNotFoundException} should cause + * immediate termination of the {@link Step}. Because it would be impossible + * for a general purpose policy to determine all the types of exceptions that + * should be skipped from those that shouldn't, a list must be passed in, with + * all of the exceptions that are 'skippable' + *

* * @author Ben Hale * @author Lucas Ward @@ -55,9 +61,8 @@ public class LimitCheckingItemSkipPolicy implements ItemSkipPolicy { private static final String SKIP = "skip"; private final int skipLimit; - + private ExceptionClassifier exceptionClassifier; - private List failurePreventingExceptions = new ArrayList(); public LimitCheckingItemSkipPolicy(int skipLimit) { this(skipLimit, new ArrayList(0)); @@ -77,41 +82,24 @@ public class LimitCheckingItemSkipPolicy implements ItemSkipPolicy { /** * Given the provided exception and skip count, determine whether or not - * processing should continue for the given exception. If the exception - * is not within the list of 'skippable exceptions', false will be returned. - * If the exception is within the list, and {@link StepExecution} skipCount - * is greater than the skipLimit, then a {@link SkipLimitExceededException} + * processing should continue for the given exception. If the exception is + * not within the list of 'skippable exceptions', false will be returned. If + * the exception is within the list, and {@link StepExecution} skipCount is + * greater than the skipLimit, then a {@link SkipLimitExceededException} * will be thrown. */ - public boolean shouldSkip(Exception ex, int skipCount){ - if(exceptionClassifier.classify(ex).equals(SKIP)){ - if(skipCount < skipLimit){ + public boolean shouldSkip(Throwable t, int skipCount) { + if (exceptionClassifier.classify(t).equals(SKIP)) { + if (skipCount < skipLimit) { return true; } - else{ - throw new SkipLimitExceededException(skipLimit, ex); + else { + throw new SkipLimitExceededException(skipLimit, t); } } - else{ + else { return false; } } - public boolean shouldFail(Throwable t) { - if(failurePreventingExceptions.contains(t)){ - return false; - } - else{ - return true; - } - } - - /** - * Set the list of exceptions that will prevent step execution from failing. - * - * @param failurePreventingExceptions - */ - public void setFailurePreventingExceptions(List failurePreventingExceptions) { - this.failurePreventingExceptions = failurePreventingExceptions; - } } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/NeverSkipItemSkipPolicy.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/NeverSkipItemSkipPolicy.java index 77b73ad9e..95e3bb074 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/NeverSkipItemSkipPolicy.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/NeverSkipItemSkipPolicy.java @@ -25,11 +25,8 @@ import org.springframework.batch.core.domain.ItemSkipPolicy; */ public class NeverSkipItemSkipPolicy implements ItemSkipPolicy{ - public boolean shouldSkip(Exception ex, int skipCount) { + public boolean shouldSkip(Throwable t, int skipCount) { return false; } - public boolean shouldFail(Throwable t) { - return true; - } } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/SimpleStepFactoryBean.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/SimpleStepFactoryBean.java index dd5b8ee88..a24f64086 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/SimpleStepFactoryBean.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/SimpleStepFactoryBean.java @@ -16,7 +16,6 @@ package org.springframework.batch.execution.step.support; import org.springframework.batch.execution.step.ItemOrientedStep; -import org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHandler; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.batch.repeat.support.RepeatTemplate; @@ -28,8 +27,6 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { private int commitInterval = 0; - private boolean alwaysSkip = false; - /** * Set the commit interval. * @@ -39,18 +36,6 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { this.commitInterval = commitInterval; } - /** - * Public setter for the flag that determines skip policy. If this flag is - * true then an exception in chunk processing will cause the item to be - * skipped and no exceptions propagated. If it is false then all exceptions - * will be propagated from the chunk and cause the step to abort. - * - * @param alwaysSkip the value to set. Default is false. - */ - public void setAlwaysSkip(boolean alwaysSkip) { - this.alwaysSkip = alwaysSkip; - } - /** * @param step * @@ -65,19 +50,5 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { step.setChunkOperations(chunkOperations); } - if (alwaysSkip) { - // If we always skip (not the default) then we are prepared to - // absorb all exceptions at the step level because the failed items - // will never re-appear after a rollback. - step.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy()); - RepeatTemplate stepOperations = new RepeatTemplate(); - stepOperations.setExceptionHandler(new SimpleLimitExceptionHandler(Integer.MAX_VALUE)); - step.setStepOperations(stepOperations); - } - else { - // This is the default in ItemOrientedStep anyway... - step.setItemSkipPolicy(new NeverSkipItemSkipPolicy()); - } - } } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/SkipLimitExceededException.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/SkipLimitExceededException.java index e5e6c61e4..edf516d0c 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/SkipLimitExceededException.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/SkipLimitExceededException.java @@ -28,8 +28,8 @@ public class SkipLimitExceededException extends InfrastructureException { private final int skipLimit; - public SkipLimitExceededException(int skipLimit, Exception ex) { - super("Skip limit of '" + skipLimit + "' exceeded", ex); + public SkipLimitExceededException(int skipLimit, Throwable t) { + super("Skip limit of '" + skipLimit + "' exceeded", t); this.skipLimit = skipLimit; } 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 new file mode 100644 index 000000000..aa4b50788 --- /dev/null +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StatefulRetryStepFactoryBean.java @@ -0,0 +1,117 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.execution.step.support; + +import org.springframework.batch.core.domain.Step; +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.retry.RetryPolicy; +import org.springframework.batch.retry.callback.ItemReaderRetryCallback; +import org.springframework.batch.retry.policy.ItemReaderRetryPolicy; +import org.springframework.batch.retry.support.RetryTemplate; + +/** + * Factory bean for step that executes its item processing with a stateful + * retry. Failed items are never skipped, but always cause a rollback. Before a + * rollback, the {@link Step} makes a record of the failed item, caching it + * under a key given by the {@link ItemKeyGenerator}. Then when it is + * re-presented by the {@link ItemReader} it is recognised and retried up to a + * limit given by the {@link RetryPolicy}. When the retry is exhausted instead + * of the item being skipped it is handled by an {@link ItemRecoverer}.
+ * + * TODO: make sure listeners are called, and add item listener callbacks to the + * recovery path. + * + * TODO: checking for null retry callback is a sucky way of determining if a + * stateful retry has been requested. + * + * @author Dave Syer + * + */ +public class StatefulRetryStepFactoryBean extends DefaultStepFactoryBean { + + private RetryPolicy retryPolicy; + + private ItemKeyGenerator itemKeyGenerator; + + /** + * Public setter for the {@link RetryPolicy}. + * @param retryPolicy the {@link RetryPolicy} to set + */ + public void setRetryPolicy(RetryPolicy retryPolicy) { + this.retryPolicy = retryPolicy; + } + + /** + * Public setter for the {@link ItemKeyGenerator} which will be used to + * cache failed items between transactions. If it is not injected but the + * reader or writer implement {@link ItemKeyGenerator}, one of those will + * be used instead (preferring the reader to the writer if both would be + * appropriate). If neither can be used, then the default will be to just + * use the item itself as a cache key. + * + * @param itemKeyGenerator the {@link ItemKeyGenerator} to set + */ + public void setItemKeyGenerator(ItemKeyGenerator itemKeyGenerator) { + this.itemKeyGenerator = itemKeyGenerator; + } + + /** + * @param step + * + */ + protected void applyConfiguration(ItemOrientedStep step) { + + // Ensure exception handler always rethrows + if (retryPolicy != null) { + super.setAlwaysSkip(true); + } + + super.applyConfiguration(step); + + if (retryPolicy != null) { + ItemReaderRetryCallback retryCallback = new ItemReaderRetryCallback(getItemReader(), getKeyGenerator(), + getItemWriter()); + ItemReaderRetryPolicy itemProviderRetryPolicy = new ItemReaderRetryPolicy(retryPolicy); + RetryTemplate template = new RetryTemplate(); + template.setRetryPolicy(itemProviderRetryPolicy); + step.setRetryOperations(template); + step.setRetryCallback(retryCallback); + } + + } + + /** + * @return an {@link ItemKeyGenerator} or null if none is found. + */ + private ItemKeyGenerator getKeyGenerator() { + + if (itemKeyGenerator != null) { + return itemKeyGenerator; + } + if (getItemReader() instanceof ItemKeyGenerator) { + return (ItemKeyGenerator) getItemReader(); + } + if (getItemWriter() instanceof ItemKeyGenerator) { + return (ItemKeyGenerator) getItemWriter(); + } + return null; + + } + +} diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/SimpleStepFactoryBeanTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/SimpleStepFactoryBeanTests.java index a8bae7db3..ce9ef4554 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/SimpleStepFactoryBeanTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/SimpleStepFactoryBeanTests.java @@ -25,7 +25,7 @@ import org.springframework.batch.core.domain.Step; */ public class SimpleStepFactoryBeanTests extends TestCase { - private AbstractStepFactoryBean factory = new KitchenSinkStepFactoryBean(); + private AbstractStepFactoryBean factory = new StatefulRetryStepFactoryBean(); public void testType() throws Exception { assertEquals(Step.class, factory.getObjectType()); diff --git a/spring-batch-samples/src/main/resources/jobs/parallelJob.xml b/spring-batch-samples/src/main/resources/jobs/parallelJob.xml index b775c6801..6622f1f33 100644 --- a/spring-batch-samples/src/main/resources/jobs/parallelJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/parallelJob.xml @@ -41,7 +41,7 @@ - + diff --git a/spring-batch-samples/src/main/resources/jobs/retrySample.xml b/spring-batch-samples/src/main/resources/jobs/retrySample.xml index c2e7c6c19..c20a3124b 100644 --- a/spring-batch-samples/src/main/resources/jobs/retrySample.xml +++ b/spring-batch-samples/src/main/resources/jobs/retrySample.xml @@ -11,8 +11,8 @@ - + @@ -23,11 +23,6 @@ value="java.lang.Exception" /> - - -