Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

OPEN - issue BATCH-404: FactoryBeans for step configuration

Browse Source 
http://jira.springframework.org/browse/BATCH-404

Remove references to KitchenSinkStepFactoryBean - use StatefulRetryStepFactoryBean for remaining sample.
This commit is contained in:
dsyer
2008-03-02 13:07:04 +00:00
parent 108c21b187
commit e5cccef3d1
13 changed files with 223 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-batch-core/src/main/java/org/springframework/batch/core/domain/ItemSkipPolicy.java
View File

@@ -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);
}

14
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java
View File

@@ -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();
}

5
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/AlwaysSkipItemSkipPolicy.java
View File

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

64
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepFactoryBean.java
View File

@@ -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());
}
}
}

172
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/KitchenSinkStepFactoryBean.java
View File

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

70
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/LimitCheckingItemSkipPolicy.java
View File

@@ -30,19 +30,25 @@ import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.io.exception.FlatFileParsingException;
/**
* <p>{@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.</p>
* <p>
* {@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.
* </p>
*
* <p>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'</p>
* <p>
* 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'
* </p>
*
* @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;
}
}

5
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/NeverSkipItemSkipPolicy.java
View File

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

29
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/SimpleStepFactoryBean.java
View File

@@ -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());
}
}
}

4
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/SkipLimitExceededException.java
View File

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

117
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StatefulRetryStepFactoryBean.java Normal file
View File

@@ -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}.<br/>
*
* 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;
}
}

2
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/SimpleStepFactoryBeanTests.java
View File

@@ -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());

2
spring-batch-samples/src/main/resources/jobs/parallelJob.xml
View File

@@ -41,7 +41,7 @@
</bean>
</property>
</bean>
<bean id="loading" parent="simpleStep" class="org.springframework.batch.execution.step.support.KitchenSinkStepFactoryBean">
<bean id="loading" parent="defaultStep">
<property name="taskExecutor">
<bean
class="org.springframework.core.task.SimpleAsyncTaskExecutor" />

9
spring-batch-samples/src/main/resources/jobs/retrySample.xml
View File

@@ -11,8 +11,8 @@
<bean id="retrySample" parent="simpleJob">
<property name="steps">
<bean id="step1" parent="simpleStep"
class="org.springframework.batch.execution.step.support.KitchenSinkStepFactoryBean">
<bean id="step1" parent="defaultStep"
class="org.springframework.batch.execution.step.support.StatefulRetryStepFactoryBean">
<property name="itemReader" ref="itemGenerator" />
<property name="itemWriter" ref="itemWriter" />
<property name="retryPolicy">
@@ -23,11 +23,6 @@
value="java.lang.Exception" />
</bean>
</property>
<property name="exceptionHandler">
<bean
class="org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHandler"
p:limit="5" />
</property>
</bean>
</property>
</bean>