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

BATCH-1913: Add step builders

Browse Source
This commit is contained in:
Dave Syer
2012-12-01 18:46:33 +00:00
parent a9140c7fac
commit f63f67c65a
43 changed files with 1948 additions and 1023 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
archetypes/simple-cli/src/main/resources/META-INF/spring/module-context.xml
View File

@@ -5,9 +5,12 @@
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd">
<description>Example job to get you started. It provides a
skeleton for a typical batch application.</description>
skeleton for
a typical batch application.
</description>
<job id="job1" xmlns="http://www.springframework.org/schema/batch" incrementer="jobParametersIncrementer">
<job id="job1" xmlns="http://www.springframework.org/schema/batch"
incrementer="jobParametersIncrementer">
<step id="step1" parent="simpleStep">
<tasklet>
<chunk reader="reader" writer="writer" />
@@ -17,13 +20,12 @@
<bean id="reader" class="example.ExampleItemReader" />
<bean id="writer" class="example.ExampleItemWriter" />
<!-- enables the functionality of JobOperator.startNextInstance(jobName) -->
<bean id="jobParametersIncrementer" class="org.springframework.batch.core.launch.support.RunIdIncrementer" />
<bean id="simpleStep"
class="org.springframework.batch.core.step.item.SimpleStepFactoryBean"
abstract="true">
<!-- enables the functionality of JobOperator.startNextInstance(jobName) -->
<bean id="jobParametersIncrementer"
class="org.springframework.batch.core.launch.support.RunIdIncrementer" />
<bean id="simpleStep" class="org.springframework.batch.core.step.factory.SimpleStepFactoryBean" abstract="true">
<property name="transactionManager" ref="transactionManager" />
<property name="jobRepository" ref="jobRepository" />
<property name="startLimit" value="100" />

2
spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanIntegrationTests.java
View File

@@ -23,7 +23,7 @@ import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;

2
spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackIntegrationTests.java
View File

@@ -26,7 +26,7 @@ import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;

2
spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/MapRepositoryFaultTolerantStepFactoryBeanRollbackTests.java
View File

@@ -22,7 +22,7 @@ import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;

2
spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/MapRepositoryFaultTolerantStepFactoryBeanTests.java
View File

@@ -20,7 +20,7 @@ import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;

569
spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java
View File

@@ -16,37 +16,41 @@
package org.springframework.batch.core.configuration.xml;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.framework.Advised;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.ItemProcessListener;
import org.springframework.batch.core.ItemReadListener;
import org.springframework.batch.core.ItemWriteListener;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.job.flow.Flow;
import org.springframework.batch.core.job.flow.FlowStep;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.batch.core.partition.PartitionHandler;
import org.springframework.batch.core.partition.support.PartitionStep;
import org.springframework.batch.core.partition.support.Partitioner;
import org.springframework.batch.core.partition.support.SimpleStepExecutionSplitter;
import org.springframework.batch.core.partition.support.StepExecutionAggregator;
import org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.AbstractStep;
import org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean;
import org.springframework.batch.core.step.builder.AbstractTaskletStepBuilder;
import org.springframework.batch.core.step.builder.FaultTolerantStepBuilder;
import org.springframework.batch.core.step.builder.FlowStepBuilder;
import org.springframework.batch.core.step.builder.JobStepBuilder;
import org.springframework.batch.core.step.builder.PartitionStepBuilder;
import org.springframework.batch.core.step.builder.SimpleStepBuilder;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.core.step.builder.StepBuilderHelper;
import org.springframework.batch.core.step.builder.TaskletStepBuilder;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.core.step.factory.SimpleStepFactoryBean;
import org.springframework.batch.core.step.item.KeyGenerator;
import org.springframework.batch.core.step.item.SimpleStepFactoryBean;
import org.springframework.batch.core.step.job.JobParametersExtractor;
import org.springframework.batch.core.step.job.JobStep;
import org.springframework.batch.core.step.skip.SkipPolicy;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.core.step.tasklet.TaskletStep;
@@ -60,7 +64,6 @@ import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.classify.BinaryExceptionClassifier;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.retry.RetryListener;
import org.springframework.retry.RetryPolicy;
@@ -74,11 +77,9 @@ import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
import org.springframework.util.Assert;
/**
* This {@link FactoryBean} is used by the batch namespace parser to create
* {@link Step} objects. Stores all of the properties that are configurable on
* the &lt;step/&gt; (and its inner &lt;tasklet/&gt;). Based on which properties
* are configured, the {@link #getObject()} method will delegate to the
* appropriate class for generating the {@link Step}.
* This {@link FactoryBean} is used by the batch namespace parser to create {@link Step} objects. Stores all of the
* properties that are configurable on the &lt;step/&gt; (and its inner &lt;tasklet/&gt;). Based on which properties are
* configured, the {@link #getObject()} method will delegate to the appropriate class for generating the {@link Step}.
*
* @author Dan Garrette
* @author Josh Long
@@ -89,8 +90,6 @@ import org.springframework.util.Assert;
*/
class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
private static final Log logger = LogFactory.getLog(StepParserStepFactoryBean.class);
//
// Step Attributes
//
@@ -109,6 +108,8 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
private PlatformTransactionManager transactionManager;
private Set<StepExecutionListener> stepExecutionListeners = new LinkedHashSet<StepExecutionListener>();
//
// Flow Elements
//
@@ -139,8 +140,6 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
//
// Tasklet Elements
//
private StepListener[] listeners;
private Collection<Class<? extends Throwable>> noRollbackExceptionClasses;
private Integer transactionTimeout;
@@ -149,10 +148,12 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
private Isolation isolation;
private Set<ChunkListener> chunkListeners = new LinkedHashSet<ChunkListener>();
//
// Chunk Attributes
//
private Integer cacheCapacity;
private int cacheCapacity = 0;
private CompletionPolicy chunkCompletionPolicy;
@@ -162,7 +163,7 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
private Boolean processorTransactional;
private Integer retryLimit;
private int retryLimit = 0;
private BackOffPolicy backOffPolicy;
@@ -191,12 +192,20 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
//
private RetryListener[] retryListeners;
private Map<Class<? extends Throwable>, Boolean> skippableExceptionClasses;
private Map<Class<? extends Throwable>, Boolean> skippableExceptionClasses = new HashMap<Class<? extends Throwable>, Boolean>();
private Map<Class<? extends Throwable>, Boolean> retryableExceptionClasses;
private Map<Class<? extends Throwable>, Boolean> retryableExceptionClasses = new HashMap<Class<? extends Throwable>, Boolean>();
private ItemStream[] streams;
private Set<ItemReadListener<I>> readListeners = new LinkedHashSet<ItemReadListener<I>>();
private Set<ItemWriteListener<O>> writeListeners = new LinkedHashSet<ItemWriteListener<O>>();
private Set<ItemProcessListener<I, O>> processListeners = new LinkedHashSet<ItemProcessListener<I, O>>();
private Set<SkipListener<I, O>> skipListeners = new LinkedHashSet<SkipListener<I, O>>();
//
// Additional
//
@@ -204,6 +213,8 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
private StepExecutionAggregator stepExecutionAggregator;
private StepListener[] listeners;
/**
* Create a {@link Step} from the configuration provided.
*
@@ -216,36 +227,23 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
validateFaultTolerantSettings();
if (isFaultTolerant()) {
FaultTolerantStepFactoryBean<I, O> fb = new FaultTolerantStepFactoryBean<I, O>();
configureSimple(fb);
configureFaultTolerant(fb);
return fb.getObject();
return createFaultTolerantStep();
}
else {
SimpleStepFactoryBean<I, O> fb = new SimpleStepFactoryBean<I, O>();
configureSimple(fb);
return fb.getObject();
return createSimpleStep();
}
}
else if (tasklet != null) {
TaskletStep ts = new TaskletStep();
configureTaskletStep(ts);
return ts;
return createTaskletStep();
}
else if (flow != null) {
FlowStep ts = new FlowStep();
configureFlowStep(ts);
return ts;
return createFlowStep();
}
else if (job != null) {
JobStep ts = new JobStep();
configureJobStep(ts);
return ts;
return createJobStep();
}
else {
PartitionStep ts = new PartitionStep();
configurePartitionStep(ts);
return ts;
return createPartitionStep();
}
}
@@ -255,215 +253,166 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
return hasChunkElement || tasklet != null;
}
private void configureAbstractStep(AbstractStep ts) {
if (name != null) {
ts.setName(name);
}
private void enhanceCommonStep(StepBuilderHelper<?> builder) {
if (allowStartIfComplete != null) {
ts.setAllowStartIfComplete(allowStartIfComplete);
}
if (jobRepository != null) {
ts.setJobRepository(jobRepository);
builder.allowStartIfComplete(allowStartIfComplete);
}
if (startLimit != null) {
ts.setStartLimit(startLimit);
builder.startLimit(startLimit);
}
if (listeners != null) {
List<StepExecutionListener> newListeners = new ArrayList<StepExecutionListener>();
for (StepListener listener : listeners) {
if (listener instanceof StepExecutionListener) {
newListeners.add((StepExecutionListener) listener);
}
}
ts.setStepExecutionListeners(newListeners.toArray(new StepExecutionListener[0]));
builder.repository(jobRepository);
builder.transactionManager(transactionManager);
for (StepExecutionListener listener : stepExecutionListeners) {
builder.listener(listener);
}
}
private void configurePartitionStep(PartitionStep ts) {
Assert.state(partitioner != null, "A Partitioner must be provided for a partition step");
configureAbstractStep(ts);
private Step createPartitionStep() {
if (partitionHandler != null) {
ts.setPartitionHandler(partitionHandler);
PartitionStepBuilder builder;
if (partitioner != null) {
builder = new StepBuilder(name).partitioner(step != null ? step.getName() : name, partitioner).step(step);
}
else {
TaskExecutorPartitionHandler partitionHandler = new TaskExecutorPartitionHandler();
partitionHandler.setStep(step);
if (taskExecutor == null) {
taskExecutor = new SyncTaskExecutor();
}
partitionHandler.setGridSize(gridSize);
partitionHandler.setTaskExecutor(taskExecutor);
ts.setPartitionHandler(partitionHandler);
builder = new StepBuilder(name).partitioner(step);
}
enhanceCommonStep(builder);
if (partitionHandler != null) {
builder.partitionHandler(partitionHandler);
}
else {
builder.gridSize(gridSize);
builder.taskExecutor(taskExecutor);
}
boolean allowStartIfComplete = this.allowStartIfComplete != null ? this.allowStartIfComplete : false;
String name = this.name;
if (step != null) {
try {
allowStartIfComplete = step.isAllowStartIfComplete();
name = step.getName();
}
catch (Exception e) {
logger.info("Ignored exception from step asking for name and allowStartIfComplete flag. "
+ "Using default from enclosing PartitionStep (" + name + "," + allowStartIfComplete + ").");
}
}
SimpleStepExecutionSplitter splitter = new SimpleStepExecutionSplitter(jobRepository, allowStartIfComplete,
name, partitioner);
ts.setStepExecutionSplitter(splitter);
if (stepExecutionAggregator != null) {
ts.setStepExecutionAggregator(stepExecutionAggregator);
}
builder.aggregator(stepExecutionAggregator);
return builder.build();
}
private Object extractTarget(Object target, Class<?> type) {
if (target instanceof Advised) {
Object source;
try {
source = ((Advised) target).getTargetSource().getTarget();
}
catch (Exception e) {
throw new IllegalStateException("Could not extract target from proxy", e);
}
if (source instanceof Advised) {
source = extractTarget(source, type);
}
if (type.isAssignableFrom(source.getClass())) {
target = source;
}
}
return target;
}
private Step createFaultTolerantStep() {
private void configureSimple(SimpleStepFactoryBean<I, O> fb) {
if (name != null) {
fb.setBeanName(name);
}
if (allowStartIfComplete != null) {
fb.setAllowStartIfComplete(allowStartIfComplete);
}
if (jobRepository != null) {
fb.setJobRepository(jobRepository);
}
if (startLimit != null) {
fb.setStartLimit(startLimit);
}
if (transactionManager != null) {
fb.setTransactionManager(transactionManager);
}
if (listeners != null) {
fb.setListeners(listeners);
}
if (transactionTimeout != null) {
fb.setTransactionTimeout(transactionTimeout);
}
if (propagation != null) {
fb.setPropagation(propagation);
}
if (isolation != null) {
fb.setIsolation(isolation);
}
FaultTolerantStepBuilder<I, O> builder = new FaultTolerantStepBuilder<I, O>(new StepBuilder(name));
if (chunkCompletionPolicy != null) {
fb.setChunkCompletionPolicy(chunkCompletionPolicy);
}
if (commitInterval != null) {
fb.setCommitInterval(commitInterval);
builder.chunk(commitInterval);
}
if (taskExecutor != null) {
fb.setTaskExecutor(taskExecutor);
}
if (throttleLimit != null) {
fb.setThrottleLimit(throttleLimit);
}
if (itemReader != null) {
fb.setItemReader(itemReader);
}
if (itemProcessor != null) {
fb.setItemProcessor(itemProcessor);
}
if (itemWriter != null) {
fb.setItemWriter(itemWriter);
enhanceTaskletStepBuilder(builder);
builder.reader(itemReader);
builder.writer(itemWriter);
builder.processor(itemProcessor);
if (processorTransactional != null && !processorTransactional) {
builder.processorNonTransactional();
}
if (streams != null) {
fb.setStreams(streams);
for (SkipListener<I, O> listener : skipListeners) {
builder.listener(listener);
}
}
private void configureFaultTolerant(FaultTolerantStepFactoryBean<I, O> fb) {
if (cacheCapacity != null) {
fb.setCacheCapacity(cacheCapacity);
}
if (readerTransactionalQueue != null) {
fb.setIsReaderTransactionalQueue(readerTransactionalQueue);
}
if (processorTransactional != null) {
fb.setProcessorTransactional(processorTransactional);
}
if (retryLimit != null) {
fb.setRetryLimit(retryLimit);
}
if (skipLimit != null) {
fb.setSkipLimit(skipLimit);
}
registerItemListeners(builder);
if (skipPolicy != null) {
fb.setSkipPolicy(skipPolicy);
builder.skipPolicy(skipPolicy);
}
if (backOffPolicy != null) {
fb.setBackOffPolicy(backOffPolicy);
}
if (retryPolicy != null) {
fb.setRetryPolicy(retryPolicy);
}
if (retryContextCache != null) {
fb.setRetryContextCache(retryContextCache);
}
if (keyGenerator != null) {
fb.setKeyGenerator(keyGenerator);
else if (skipLimit!=null) {
builder.skipLimit(skipLimit);
for (Class<? extends Throwable> type : skippableExceptionClasses.keySet()) {
if (skippableExceptionClasses.get(type)) {
builder.skip(type);
}
else {
builder.noSkip(type);
}
}
}
if (retryListeners != null) {
fb.setRetryListeners(retryListeners);
for (RetryListener listener : retryListeners) {
builder.listener(listener);
}
}
if (skippableExceptionClasses != null) {
fb.setSkippableExceptionClasses(skippableExceptionClasses);
if (retryContextCache == null && cacheCapacity > 0) {
retryContextCache = new MapRetryContextCache(cacheCapacity);
}
if (retryableExceptionClasses != null) {
fb.setRetryableExceptionClasses(retryableExceptionClasses);
builder.retryContextCache(retryContextCache);
builder.keyGenerator(keyGenerator);
if (retryPolicy != null) {
builder.retryPolicy(retryPolicy);
}
else {
builder.retryLimit(retryLimit);
builder.backOffPolicy(backOffPolicy);
for (Class<? extends Throwable> type : retryableExceptionClasses.keySet()) {
if (retryableExceptionClasses.get(type)) {
builder.retry(type);
}
else {
builder.noRetry(type);
}
}
}
if (noRollbackExceptionClasses != null) {
fb.setNoRollbackExceptionClasses(noRollbackExceptionClasses);
for (Class<? extends Throwable> type : noRollbackExceptionClasses) {
builder.noRollback(type);
}
}
return builder.build();
}
private void registerItemListeners(SimpleStepBuilder<I, O> builder) {
for (ItemReadListener<I> listener : readListeners) {
builder.listener(listener);
}
for (ItemWriteListener<O> listener : writeListeners) {
builder.listener(listener);
}
for (ItemProcessListener<I, O> listener : processListeners) {
builder.listener(listener);
}
}
@SuppressWarnings("serial")
private void configureTaskletStep(TaskletStep ts) {
configureAbstractStep(ts);
if (listeners != null) {
List<ChunkListener> newListeners = new ArrayList<ChunkListener>();
for (StepListener listener : listeners) {
if (listener instanceof ChunkListener) {
newListeners.add((ChunkListener) listener);
}
}
ts.setChunkListeners(newListeners.toArray(new ChunkListener[0]));
@SuppressWarnings("unchecked")
private Step createSimpleStep() {
@SuppressWarnings("rawtypes")
SimpleStepBuilder builder = new SimpleStepBuilder(new StepBuilder(name));
if (commitInterval != null) {
builder.chunk(commitInterval);
}
if (tasklet != null) {
ts.setTasklet(tasklet);
enhanceTaskletStepBuilder(builder);
registerItemListeners(builder);
builder.reader(itemReader);
builder.writer(itemWriter);
builder.processor(itemProcessor);
builder.completionPolicy(chunkCompletionPolicy);
return builder.build();
}
private TaskletStep createTaskletStep() {
TaskletStepBuilder builder = new StepBuilder(name).tasklet(tasklet);
enhanceTaskletStepBuilder(builder);
return builder.build();
}
private void enhanceTaskletStepBuilder(AbstractTaskletStepBuilder<?> builder) {
enhanceCommonStep(builder);
for (ChunkListener listener : chunkListeners) {
builder.listener(listener);
}
if (taskExecutor != null) {
TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate();
repeatTemplate.setTaskExecutor(taskExecutor);
if (throttleLimit != null) {
repeatTemplate.setThrottleLimit(throttleLimit);
}
ts.setStepOperations(repeatTemplate);
}
if (transactionManager != null) {
ts.setTransactionManager(transactionManager);
builder.taskExecutor(taskExecutor);
if (throttleLimit != null) {
builder.throttleLimit(throttleLimit);
}
builder.transactionManager(transactionManager);
if (transactionTimeout != null || propagation != null || isolation != null
|| noRollbackExceptionClasses != null) {
DefaultTransactionAttribute attribute = new DefaultTransactionAttribute();
@@ -479,39 +428,35 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
Collection<Class<? extends Throwable>> exceptions = noRollbackExceptionClasses == null ? new HashSet<Class<? extends Throwable>>()
: noRollbackExceptionClasses;
final BinaryExceptionClassifier classifier = new BinaryExceptionClassifier(exceptions, false);
ts.setTransactionAttribute(new DefaultTransactionAttribute(attribute) {
builder.transactionAttribute(new DefaultTransactionAttribute(attribute) {
@Override
public boolean rollbackOn(Throwable ex) {
return classifier.classify(ex);
}
});
}
if (streams != null) {
for (ItemStream stream : streams) {
builder.stream(stream);
}
}
}
@SuppressWarnings("serial")
private void configureFlowStep(FlowStep ts) {
configureAbstractStep(ts);
if (flow != null) {
ts.setFlow(flow);
}
private Step createFlowStep() {
FlowStepBuilder builder = new StepBuilder(name).flow(flow);
enhanceCommonStep(builder);
return builder.build();
}
@SuppressWarnings("serial")
private void configureJobStep(JobStep ts) throws Exception {
configureAbstractStep(ts);
if (job != null) {
ts.setJob(job);
}
if (jobParametersExtractor != null) {
ts.setJobParametersExtractor(jobParametersExtractor);
}
if (jobLauncher == null) {
SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
jobLauncher.setJobRepository(jobRepository);
jobLauncher.afterPropertiesSet();
this.jobLauncher = jobLauncher;
}
ts.setJobLauncher(jobLauncher);
private Step createJobStep() throws Exception {
JobStepBuilder builder = new StepBuilder(name).job(job);
enhanceCommonStep(builder);
builder.parametersExtractor(jobParametersExtractor);
builder.launcher(jobLauncher);
return builder.build();
}
private void validateFaultTolerantSettings() {
@@ -526,14 +471,12 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
}
/**
* Check if a field is present then a second is also. If the
* twoWayDependency flag is set then the opposite must also be true: if the
* second value is present, the first must also be.
* Check if a field is present then a second is also. If the twoWayDependency flag is set then the opposite must
* also be true: if the second value is present, the first must also be.
*
* @param dependentName the name of the first field
* @param dependentValue the value of the first field
* @param name the name of the other field (which should be absent if the
* first is present)
* @param name the name of the other field (which should be absent if the first is present)
* @param value the value of the other field
* @param twoWayDependency true if both depend on each other
* @throws IllegalArgumentException if either condition is violated
@@ -560,6 +503,12 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
if (o instanceof Integer) {
return isPositive((Integer) o);
}
if (o instanceof Collection) {
return !((Collection<?>) o).isEmpty();
}
if (o instanceof Map) {
return !((Map<?, ?>) o).isEmpty();
}
return o != null;
}
@@ -589,8 +538,7 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
// =========================================================
/**
* Set the bean name property, which will become the name of the
* {@link Step} when it is created.
* Set the bean name property, which will become the name of the {@link Step} when it is created.
*
* @see org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang.String)
*/
@@ -678,8 +626,8 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
// =========================================================
/**
* Public setter for the flag to indicate that the step should be replayed
* on a restart, even if successful the first time.
* Public setter for the flag to indicate that the step should be replayed on a restart, even if successful the
* first time.
*
* @param allowStartIfComplete the shouldAllowStartIfComplete to set
*/
@@ -741,19 +689,47 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
// =========================================================
/**
* The listeners to inject into the {@link Step}. Any instance of
* {@link StepListener} can be used, and will then receive callbacks at the
* appropriate stage in the step.
* The listeners to inject into the {@link Step}. Any instance of {@link StepListener} can be used, and will then
* receive callbacks at the appropriate stage in the step.
*
* @param listeners an array of listeners
*/
public void setListeners(StepListener[] listeners) {
this.listeners = listeners;
this.listeners = listeners; // useful for testing
for (StepListener listener : listeners) {
if (listener instanceof SkipListener) {
@SuppressWarnings("unchecked")
SkipListener<I, O> skipListener = (SkipListener<I, O>) listener;
skipListeners.add(skipListener);
}
if (listener instanceof StepExecutionListener) {
StepExecutionListener stepExecutionListener = (StepExecutionListener) listener;
stepExecutionListeners.add(stepExecutionListener);
}
if (listener instanceof ChunkListener) {
ChunkListener chunkListener = (ChunkListener) listener;
chunkListeners.add(chunkListener);
}
if (listener instanceof ItemReadListener) {
@SuppressWarnings("unchecked")
ItemReadListener<I> readListener = (ItemReadListener<I>) listener;
readListeners.add(readListener);
}
if (listener instanceof ItemWriteListener) {
@SuppressWarnings("unchecked")
ItemWriteListener<O> writeListener = (ItemWriteListener<O>) listener;
writeListeners.add(writeListener);
}
if (listener instanceof ItemProcessListener) {
@SuppressWarnings("unchecked")
ItemProcessListener<I, O> processListener = (ItemProcessListener<I, O>) listener;
processListeners.add(processListener);
}
}
}
/**
* Exception classes that may not cause a rollback if encountered in the
* right place.
* Exception classes that may not cause a rollback if encountered in the right place.
*
* @param noRollbackExceptionClasses the noRollbackExceptionClasses to set
*/
@@ -796,8 +772,8 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
}
/**
* A retry policy to apply when exceptions occur. If this is specified then
* the retry limit and retryable exceptions will be ignored.
* A retry policy to apply when exceptions occur. If this is specified then the retry limit and retryable exceptions
* will be ignored.
*
* @param retryPolicy the {@link RetryPolicy} to set
*/
@@ -813,9 +789,8 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
}
/**
* A key generator that can be used to compare items with previously
* recorded items in a retry. Only used if the reader is a transactional
* queue.
* A key generator that can be used to compare items with previously recorded items in a retry. Only used if the
* reader is a transactional queue.
*
* @param keyGenerator the {@link KeyGenerator} to set
*/
@@ -828,28 +803,24 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
// =========================================================
/**
* Public setter for the capacity of the cache in the retry policy. If more
* items than this fail without being skipped or recovered an exception will
* be thrown. This is to guard against inadvertent infinite loops generated
* Public setter for the capacity of the cache in the retry policy. If more items than this fail without being
* skipped or recovered an exception will be thrown. This is to guard against inadvertent infinite loops generated
* by item identity problems.<br/>
* <p/>
* The default value should be high enough and more for most purposes. To
* breach the limit in a single-threaded step typically you have to have
* this many failures in a single transaction. Defaults to the value in the
* The default value should be high enough and more for most purposes. To breach the limit in a single-threaded step
* typically you have to have this many failures in a single transaction. Defaults to the value in the
* {@link MapRetryContextCache}.<br/>
*
* @param cacheCapacity the cache capacity to set (greater than 0 else
* ignored)
* @param cacheCapacity the cache capacity to set (greater than 0 else ignored)
*/
public void setCacheCapacity(int cacheCapacity) {
this.cacheCapacity = cacheCapacity;
}
/**
* Public setter for the {@link CompletionPolicy} applying to the chunk
* level. A transaction will be committed when this policy decides to
* complete. Defaults to a {@link SimpleCompletionPolicy} with chunk size
* equal to the commitInterval property.
* Public setter for the {@link CompletionPolicy} applying to the chunk level. A transaction will be committed when
* this policy decides to complete. Defaults to a {@link SimpleCompletionPolicy} with chunk size equal to the
* commitInterval property.
*
* @param chunkCompletionPolicy the chunkCompletionPolicy to set
*/
@@ -858,8 +829,7 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
}
/**
* Set the commit interval. Either set this or the chunkCompletionPolicy but
* not both.
* Set the commit interval. Either set this or the chunkCompletionPolicy but not both.
*
* @param commitInterval 1 by default
*/
@@ -868,9 +838,8 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
}
/**
* Flag to signal that the reader is transactional (usually a JMS consumer)
* so that items are re-presented after a rollback. The default is false and
* readers are assumed to be forward-only.
* Flag to signal that the reader is transactional (usually a JMS consumer) so that items are re-presented after a
* rollback. The default is false and readers are assumed to be forward-only.
*
* @param isReaderTransactionalQueue the value of the flag
*/
@@ -879,10 +848,8 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
}
/**
* Flag to signal that the processor is transactional, in which case it
* should be called for every item in every transaction. If false then we
* can cache the processor results between transactions in the case of a
* rollback.
* Flag to signal that the processor is transactional, in which case it should be called for every item in every
* transaction. If false then we can cache the processor results between transactions in the case of a rollback.
*
* @param processorTransactional the value to set
*/
@@ -891,9 +858,8 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
}
/**
* Public setter for the retry limit. Each item can be retried up to this
* limit. Note this limit includes the initial attempt to process the item,
* therefore <code>retryLimit == 1</code> by default.
* Public setter for the retry limit. Each item can be retried up to this limit. Note this limit includes the
* initial attempt to process the item, therefore <code>retryLimit == 1</code> by default.
*
* @param retryLimit the retry limit to set, must be greater or equal to 1.
*/
@@ -902,11 +868,9 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
}
/**
* Public setter for a limit that determines skip policy. If this value is
* positive then an exception in chunk processing will cause the item to be
* skipped and no exception propagated until the limit is reached. If it is
* zero then all exceptions will be propagated from the chunk and cause the
* step to abort.
* Public setter for a limit that determines skip policy. If this value is positive then an exception in chunk
* processing will cause the item to be skipped and no exception propagated until the limit is reached. If it is
* zero then all exceptions will be propagated from the chunk and cause the step to abort.
*
* @param skipLimit the value to set. Default is 0 (never skip).
*/
@@ -915,8 +879,7 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
}
/**
* Public setter for a skip policy. If this value is set then the skip limit
* and skippable exceptions are ignored.
* Public setter for a skip policy. If this value is set then the skip limit and skippable exceptions are ignored.
*
* @param skipPolicy the {@link SkipPolicy} to set
*/
@@ -925,8 +888,8 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
}
/**
* Public setter for the {@link TaskExecutor}. If this is set, then it will
* be used to execute the chunk processing inside the {@link Step}.
* 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
*/
@@ -935,10 +898,8 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
}
/**
* Public setter for the throttle limit. This limits the number of tasks
* queued for concurrent processing to prevent thread pools from being
* overwhelmed. Defaults to
* {@link TaskExecutorRepeatTemplate#DEFAULT_THROTTLE_LIMIT}.
* Public setter for the throttle limit. This limits the number of tasks queued for concurrent processing to prevent
* thread pools from being overwhelmed. Defaults to {@link TaskExecutorRepeatTemplate#DEFAULT_THROTTLE_LIMIT}.
*
* @param throttleLimit the throttle limit to set.
*/
@@ -981,9 +942,8 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
}
/**
* Public setter for exception classes that when raised won't crash the job
* but will result in transaction rollback and the item which handling
* caused the exception will be skipped.
* Public setter for exception classes that when raised won't crash the job but will result in transaction rollback
* and the item which handling caused the exception will be skipped.
*
* @param exceptionClasses
*/
@@ -1001,9 +961,8 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
}
/**
* The streams to inject into the {@link Step}. Any instance of
* {@link ItemStream} can be used, and will then receive callbacks at the
* appropriate stage in the step.
* The streams to inject into the {@link Step}. Any instance of {@link ItemStream} can be used, and will then
* receive callbacks at the appropriate stage in the step.
*
* @param streams an array of listeners
*/

5
spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowStep.java
View File

@@ -55,8 +55,11 @@ public class FlowStep extends AbstractStep {
*/
@Override
public void afterPropertiesSet() throws Exception {
super.afterPropertiesSet();
Assert.state(flow != null, "A Flow must be provided");
if (getName()==null) {
setName(flow.getName());
}
super.afterPropertiesSet();
}
/**

49
spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java
View File

@@ -39,8 +39,8 @@ import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
/**
* A {@link Step} implementation that provides common behavior to subclasses,
* including registering and calling listeners.
* A {@link Step} implementation that provides common behavior to subclasses, including registering and calling
* listeners.
*
* @author Dave Syer
* @author Ben Hale
@@ -68,7 +68,8 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
}
public void afterPropertiesSet() throws Exception {
Assert.notNull(jobRepository, "JobRepository is mandatory");
Assert.state(name != null, "A Step must have a name");
Assert.state(jobRepository != null, "JobRepository is mandatory");
}
public String getName() {
@@ -76,8 +77,7 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
}
/**
* Set the name property. Always overrides the default value if this object
* is a Spring bean.
* Set the name property. Always overrides the default value if this object is a Spring bean.
*
* @see #setBeanName(java.lang.String)
*/
@@ -86,11 +86,9 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
}
/**
* Set the name property if it is not already set. Because of the order of
* the callbacks in a Spring container the name property will be set first
* if it is present. Care is needed with bean definition inheritance - if a
* parent bean has a name, then its children need an explicit name as well,
* otherwise they will not be unique.
* Set the name property if it is not already set. Because of the order of the callbacks in a Spring container the
* name property will be set first if it is present. Care is needed with bean definition inheritance - if a parent
* bean has a name, then its children need an explicit name as well, otherwise they will not be unique.
*
* @see org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang.String)
*/
@@ -118,8 +116,8 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
}
/**
* Public setter for flag that determines whether the step should start
* again if it is already complete. Defaults to false.
* Public setter for flag that determines whether the step should start again if it is already complete. Defaults to
* false.
*
* @param allowStartIfComplete the value of the flag to set
*/
@@ -137,9 +135,8 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
}
/**
* Extension point for subclasses to execute business logic. Subclasses
* should set the {@link ExitStatus} on the {@link StepExecution} before
* returning.
* Extension point for subclasses to execute business logic. Subclasses should set the {@link ExitStatus} on the
* {@link StepExecution} before returning.
*
* @param stepExecution the current step context
* @throws Exception
@@ -147,9 +144,8 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
protected abstract void doExecute(StepExecution stepExecution) throws Exception;
/**
* Extension point for subclasses to provide callbacks to their
* collaborators at the beginning of a step, to open or acquire resources.
* Does nothing by default.
* Extension point for subclasses to provide callbacks to their collaborators at the beginning of a step, to open or
* acquire resources. Does nothing by default.
*
* @param ctx the {@link ExecutionContext} to use
* @throws Exception
@@ -158,9 +154,8 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
}
/**
* Extension point for subclasses to provide callbacks to their
* collaborators at the end of a step (right at the end of the finally
* block), to close or release resources. Does nothing by default.
* Extension point for subclasses to provide callbacks to their collaborators at the end of a step (right at the end
* of the finally block), to close or release resources. Does nothing by default.
*
* @param ctx the {@link ExecutionContext} to use
* @throws Exception
@@ -169,9 +164,8 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
}
/**
* Template method for step execution logic - calls abstract methods for
* resource initialization ({@link #open(ExecutionContext)}), execution
* logic ({@link #doExecute(StepExecution)}) and resource closing (
* Template method for step execution logic - calls abstract methods for resource initialization (
* {@link #open(ExecutionContext)}), execution logic ({@link #doExecute(StepExecution)}) and resource closing (
* {@link #close(ExecutionContext)}).
*/
public final void execute(StepExecution stepExecution) throws JobInterruptedException,
@@ -287,8 +281,7 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
}
/**
* Register a step listener for callbacks at the appropriate stages in a
* step execution.
* Register a step listener for callbacks at the appropriate stages in a step execution.
*
* @param listener a {@link StepExecutionListener}
*/
@@ -332,8 +325,8 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
}
/**
* Default mapping from throwable to {@link ExitStatus}. Clients can modify
* the exit code using a {@link StepExecutionListener}.
* Default mapping from throwable to {@link ExitStatus}. Clients can modify the exit code using a
* {@link StepExecutionListener}.
*
* @param ex the cause of the failure
* @return an {@link ExitStatus}

136
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/AbstractTaskletStepBuilder.java Normal file
View File

@@ -0,0 +1,136 @@
package org.springframework.batch.core.step.builder;
import java.util.LinkedHashSet;
import java.util.Set;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.repeat.RepeatOperations;
import org.springframework.batch.repeat.exception.DefaultExceptionHandler;
import org.springframework.batch.repeat.exception.ExceptionHandler;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.transaction.interceptor.TransactionAttribute;
public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBuilder<B>> extends
StepBuilderHelper<AbstractTaskletStepBuilder<B>> {
private Set<ChunkListener> listeners = new LinkedHashSet<ChunkListener>();
private RepeatOperations stepOperations;
private TransactionAttribute transactionAttribute;
private Set<ItemStream> streams = new LinkedHashSet<ItemStream>();
private ExceptionHandler exceptionHandler = new DefaultExceptionHandler();
private int throttleLimit = TaskExecutorRepeatTemplate.DEFAULT_THROTTLE_LIMIT;
private TaskExecutor taskExecutor;
public AbstractTaskletStepBuilder(StepBuilderHelper<?> parent) {
super(parent);
}
protected abstract Tasklet createTasklet();
public TaskletStep build() {
TaskletStep step = new TaskletStep(getName());
super.enhance(step);
step.setChunkListeners(listeners.toArray(new ChunkListener[0]));
if (transactionAttribute != null) {
step.setTransactionAttribute(transactionAttribute);
}
if (stepOperations == null) {
stepOperations = new RepeatTemplate();
if (taskExecutor != null) {
TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate();
repeatTemplate.setTaskExecutor(taskExecutor);
repeatTemplate.setThrottleLimit(throttleLimit);
stepOperations = repeatTemplate;
}
((RepeatTemplate) stepOperations).setExceptionHandler(exceptionHandler);
}
step.setStepOperations(stepOperations);
step.setTasklet(createTasklet());
step.setStreams(getStreams());
try {
step.afterPropertiesSet();
}
catch (Exception e) {
throw new IllegalStateException(e);
}
return step;
}
public AbstractTaskletStepBuilder<B> listener(ChunkListener listener) {
listeners.add(listener);
return this;
}
public AbstractTaskletStepBuilder<B> stream(ItemStream stream) {
streams.add(stream);
return this;
}
public AbstractTaskletStepBuilder<B> taskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
return this;
}
public AbstractTaskletStepBuilder<B> throttleLimit(int throttleLimit) {
this.throttleLimit = throttleLimit;
return this;
}
public AbstractTaskletStepBuilder<B> exceptionHandler(ExceptionHandler exceptionHandler) {
this.exceptionHandler = exceptionHandler;
return this;
}
public AbstractTaskletStepBuilder<B> stepOperations(RepeatOperations repeatTemplate) {
this.stepOperations = repeatTemplate;
return this;
}
public AbstractTaskletStepBuilder<B> transactionAttribute(TransactionAttribute transactionAttribute) {
this.transactionAttribute = transactionAttribute;
return this;
}
protected ItemStream[] getStreams() {
return streams.toArray(new ItemStream[0]);
}
protected RepeatOperations getStepOperations() {
return stepOperations;
}
protected ExceptionHandler getExceptionHandler() {
return exceptionHandler;
}
protected boolean concurrent() {
boolean concurrent = taskExecutor != null && !(taskExecutor instanceof SyncTaskExecutor);
return concurrent;
}
}

592
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java → spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java Executable file → Normal file
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2007 the original author or authors.
* Copyright 2006-2011 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.
@@ -13,24 +13,31 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.step.item;
package org.springframework.batch.core.step.builder;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.springframework.classify.BinaryExceptionClassifier;
import org.springframework.classify.Classifier;
import org.springframework.classify.SubclassClassifier;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.listener.StepListenerFactoryBean;
import org.springframework.batch.core.step.FatalStepExecutionException;
import org.springframework.batch.core.step.item.BatchRetryTemplate;
import org.springframework.batch.core.step.item.ChunkMonitor;
import org.springframework.batch.core.step.item.ChunkOrientedTasklet;
import org.springframework.batch.core.step.item.FaultTolerantChunkProcessor;
import org.springframework.batch.core.step.item.FaultTolerantChunkProvider;
import org.springframework.batch.core.step.item.ForceRollbackForWriteSkipException;
import org.springframework.batch.core.step.item.KeyGenerator;
import org.springframework.batch.core.step.item.SimpleRetryExceptionHandler;
import org.springframework.batch.core.step.skip.CompositeSkipPolicy;
import org.springframework.batch.core.step.skip.ExceptionClassifierSkipPolicy;
import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
@@ -40,11 +47,14 @@ import org.springframework.batch.core.step.skip.SkipLimitExceededException;
import org.springframework.batch.core.step.skip.SkipListenerFailedException;
import org.springframework.batch.core.step.skip.SkipPolicy;
import org.springframework.batch.core.step.skip.SkipPolicyFailedException;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.repeat.RepeatOperations;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.classify.BinaryExceptionClassifier;
import org.springframework.classify.Classifier;
import org.springframework.classify.SubclassClassifier;
import org.springframework.retry.ExhaustedRetryException;
import org.springframework.retry.RetryException;
import org.springframework.retry.RetryListener;
@@ -52,60 +62,29 @@ import org.springframework.retry.RetryPolicy;
import org.springframework.retry.backoff.BackOffPolicy;
import org.springframework.retry.policy.CompositeRetryPolicy;
import org.springframework.retry.policy.ExceptionClassifierRetryPolicy;
import org.springframework.retry.policy.MapRetryContextCache;
import org.springframework.retry.policy.NeverRetryPolicy;
import org.springframework.retry.policy.RetryContextCache;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.transaction.TransactionException;
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
import org.springframework.transaction.interceptor.TransactionAttribute;
import org.springframework.util.Assert;
/**
* Factory bean for step that provides options for configuring skip behaviour.
* User can set {@link #setSkipLimit(int)} to set how many exceptions of
* {@link #setSkippableExceptionClasses(Collection)} types are tolerated.
* {@link #setFatalExceptionClasses(Collection)} will cause immediate
* termination of job - they are treated as higher priority than
* {@link #setSkippableExceptionClasses(Collection)}, so the two lists don't
* need to be exclusive.
*
* Skippable exceptions on write will by default cause transaction rollback - to
* avoid rollback for specific exception class include it in the transaction
* attribute as "no rollback for".
*
* @see SimpleStepFactoryBean
*
* @author Dave Syer
* @author Robert Kasanicky
* @author Morten Andersen-Gott
*
*/
public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S> {
public class FaultTolerantStepBuilder<I, O> extends SimpleStepBuilder<I, O> {
private Map<Class<? extends Throwable>, Boolean> skippableExceptionClasses = new HashMap<Class<? extends Throwable>, Boolean>();
private ChunkMonitor chunkMonitor = new ChunkMonitor();
private Collection<Class<? extends Throwable>> nonSkippableExceptionClasses = new HashSet<Class<? extends Throwable>>();
private Collection<Class<? extends Throwable>> noRollbackExceptionClasses = new HashSet<Class<? extends Throwable>>();
private Map<Class<? extends Throwable>, Boolean> retryableExceptionClasses = new HashMap<Class<? extends Throwable>, Boolean>();
private Collection<Class<? extends Throwable>> nonRetryableExceptionClasses = new HashSet<Class<? extends Throwable>>();
private int cacheCapacity = 0;
private boolean streamIsReader;
private int retryLimit = 0;
private int skipLimit = 0;
private SkipPolicy skipPolicy;
private BackOffPolicy backOffPolicy;
private RetryListener[] retryListeners;
private Set<RetryListener> retryListeners = new LinkedHashSet<RetryListener>();
private RetryPolicy retryPolicy;
@@ -113,182 +92,235 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
private KeyGenerator keyGenerator;
private ChunkMonitor chunkMonitor = new ChunkMonitor();
private Collection<Class<? extends Throwable>> noRollbackExceptionClasses = new LinkedHashSet<Class<? extends Throwable>>();
private Map<Class<? extends Throwable>, Boolean> skippableExceptionClasses = new HashMap<Class<? extends Throwable>, Boolean>();
private Collection<Class<? extends Throwable>> nonSkippableExceptionClasses = new HashSet<Class<? extends Throwable>>();
private Map<Class<? extends Throwable>, Boolean> retryableExceptionClasses = new HashMap<Class<? extends Throwable>, Boolean>();
private Collection<Class<? extends Throwable>> nonRetryableExceptionClasses = new HashSet<Class<? extends Throwable>>();
private Set<SkipListener<? super I, ? super O>> skipListeners = new LinkedHashSet<SkipListener<? super I, ? super O>>();
private int skipLimit = 0;
private SkipPolicy skipPolicy;
private boolean processorTransactional = true;
/**
* The {@link KeyGenerator} to use to identify failed items across rollback.
* Not used in the case of the
* {@link #setIsReaderTransactionalQueue(boolean) transactional queue flag}
* being false (the default).
*
* @param keyGenerator
* the {@link KeyGenerator} to set
*/
public void setKeyGenerator(KeyGenerator keyGenerator) {
public FaultTolerantStepBuilder(StepBuilderHelper<?> parent) {
super(parent);
}
@Override
protected Tasklet createTasklet() {
Assert.state(getReader() != null, "ItemReader must be provided");
Assert.state(getProcessor() != null || getWriter() != null, "ItemWriter or ItemProcessor must be provided");
addSpecialExceptions();
registerSkipListeners();
FaultTolerantChunkProvider<I> chunkProvider = createChunkProvider();
FaultTolerantChunkProcessor<I, O> chunkProcessor = createChunkProcessor();
ChunkOrientedTasklet<I> tasklet = new ChunkOrientedTasklet<I>(chunkProvider, chunkProcessor);
tasklet.setBuffering(!isReaderTransactionalQueue());
return tasklet;
}
public FaultTolerantStepBuilder<I, O> listener(SkipListener<? super I, ? super O> listener) {
skipListeners.add(listener);
return this;
}
@Override
public FaultTolerantStepBuilder<I, O> listener(ChunkListener listener) {
super.listener(new TerminateOnExceptionChunkListenerDelegate(listener));
return this;
}
@Override
public AbstractTaskletStepBuilder<SimpleStepBuilder<I, O>> transactionAttribute(
TransactionAttribute transactionAttribute) {
return super.transactionAttribute(getTransactionAttribute(transactionAttribute));
}
public FaultTolerantStepBuilder<I, O> listener(RetryListener listener) {
retryListeners.add(listener);
return this;
}
public FaultTolerantStepBuilder<I, O> keyGenerator(KeyGenerator keyGenerator) {
this.keyGenerator = keyGenerator;
return this;
}
/**
* Setter for the retry policy. If this is specified the other retry
* properties are ignored (retryLimit, backOffPolicy,
* retryableExceptionClasses).
*
* @param retryPolicy
* a stateless {@link RetryPolicy}
*/
public void setRetryPolicy(RetryPolicy retryPolicy) {
this.retryPolicy = retryPolicy;
}
/**
* Public setter for the retry limit. Each item can be retried up to this
* limit. Note this limit includes the initial attempt to process the item,
* therefore <code>retryLimit == 1</code> by default.
*
* @param retryLimit
* the retry limit to set, must be greater or equal to 1.
*/
public void setRetryLimit(int retryLimit) {
public FaultTolerantStepBuilder<I, O> retryLimit(int retryLimit) {
this.retryLimit = retryLimit;
return this;
}
/**
* Public setter for the capacity of the cache in the retry policy. If more
* items than this fail without being skipped or recovered an exception will
* be thrown. This is to guard against inadvertent infinite loops generated
* by item identity problems.<br/>
*
* The default value should be high enough and more for most purposes. To
* breach the limit in a single-threaded step typically you have to have
* this many failures in a single transaction. Defaults to the value in the
* {@link MapRetryContextCache}.<br/>
*
* This property is ignored if the
* {@link #setRetryContextCache(RetryContextCache)} is set directly.
*
* @param cacheCapacity
* the cache capacity to set (greater than 0 else ignored)
*/
public void setCacheCapacity(int cacheCapacity) {
this.cacheCapacity = cacheCapacity;
public FaultTolerantStepBuilder<I, O> retryPolicy(RetryPolicy retryPolicy) {
this.retryPolicy = retryPolicy;
return this;
}
/**
* Override the default retry context cache for retry of chunk processing.
* If this property is set then {@link #setCacheCapacity(int)} is ignored.
*
* @param retryContextCache
* the {@link RetryContextCache} to set
*/
public void setRetryContextCache(RetryContextCache retryContextCache) {
this.retryContextCache = retryContextCache;
}
/**
* Public setter for the retryable exceptions classifier map (from throwable
* class to boolean, true is retryable).
*
* @param retryableExceptionClasses
* the retryableExceptionClasses to set
*/
public void setRetryableExceptionClasses(Map<Class<? extends Throwable>, Boolean> retryableExceptionClasses) {
this.retryableExceptionClasses = retryableExceptionClasses;
}
/**
* Public setter for the {@link BackOffPolicy}.
*
* @param backOffPolicy
* the {@link BackOffPolicy} to set
*/
public void setBackOffPolicy(BackOffPolicy backOffPolicy) {
public FaultTolerantStepBuilder<I, O> backOffPolicy(BackOffPolicy backOffPolicy) {
this.backOffPolicy = backOffPolicy;
return this;
}
/**
* Public setter for the {@link RetryListener}s.
*
* @param retryListeners
* the {@link RetryListener}s to set
*/
public void setRetryListeners(RetryListener... retryListeners) {
this.retryListeners = retryListeners;
public FaultTolerantStepBuilder<I, O> retryContextCache(RetryContextCache retryContextCache) {
this.retryContextCache = retryContextCache;
return this;
}
/**
* A limit that determines skip policy. If this value is positive then an
* exception in chunk processing will cause the item to be skipped and no
* exception propagated until the limit is reached. If it is zero then all
* exceptions will be propagated from the chunk and cause the step to abort.
*
* @param skipLimit
* the value to set. Default is 0 (never skip).
*/
public void setSkipLimit(int skipLimit) {
public FaultTolerantStepBuilder<I, O> skipLimit(int skipLimit) {
this.skipLimit = skipLimit;
return this;
}
/**
* A {@link SkipPolicy} that determines the outcome of an exception when
* processing an item. Overrides the {@link #setSkipLimit(int) skipLimit}.
* The {@link #setSkippableExceptionClasses(Map) skippableExceptionClasses}
* are also ignored if this is set.
*
* @param skipPolicy
* the {@link SkipPolicy} to set
*/
public void setSkipPolicy(SkipPolicy skipPolicy) {
public FaultTolerantStepBuilder<I, O> skipPolicy(SkipPolicy skipPolicy) {
this.skipPolicy = skipPolicy;
return this;
}
public FaultTolerantStepBuilder<I, O> noRollback(Class<? extends Throwable> type) {
noRollbackExceptionClasses.add(type);
return this;
}
public FaultTolerantStepBuilder<I, O> noRetry(Class<? extends Throwable> type) {
retryableExceptionClasses.put(type, false);
return this;
}
public FaultTolerantStepBuilder<I, O> retry(Class<? extends Throwable> type) {
retryableExceptionClasses.put(type, true);
return this;
}
public FaultTolerantStepBuilder<I, O> noSkip(Class<? extends Throwable> type) {
skippableExceptionClasses.put(type, false);
return this;
}
public FaultTolerantStepBuilder<I, O> skip(Class<? extends Throwable> type) {
skippableExceptionClasses.put(type, true);
return this;
}
public FaultTolerantStepBuilder<I, O> processorNonTransactional() {
this.processorTransactional = false;
return this;
}
@Override
public AbstractTaskletStepBuilder<SimpleStepBuilder<I, O>> stream(ItemStream stream) {
if (stream instanceof ItemReader<?>) {
if (!streamIsReader) {
streamIsReader = true;
super.stream(chunkMonitor);
}
// In cases where multiple nested item readers are registered,
// they all want to get the open() and close() callbacks.
chunkMonitor.registerItemStream(stream);
} else {
super.stream(stream);
}
return this;
}
private FaultTolerantChunkProvider<I> createChunkProvider() {
SkipPolicy readSkipPolicy = createSkipPolicy();
readSkipPolicy = getFatalExceptionAwareProxy(readSkipPolicy);
FaultTolerantChunkProvider<I> chunkProvider = new FaultTolerantChunkProvider<I>(getReader(),
createChunkOperations());
chunkProvider.setMaxSkipsOnRead(Math.max(getChunkSize(), FaultTolerantChunkProvider.DEFAULT_MAX_SKIPS_ON_READ));
chunkProvider.setSkipPolicy(readSkipPolicy);
chunkProvider.setRollbackClassifier(getRollbackClassifier());
ArrayList<StepListener> listeners = new ArrayList<StepListener>(getItemListeners());
listeners.addAll(skipListeners);
chunkProvider.setListeners(listeners);
return chunkProvider;
}
private FaultTolerantChunkProcessor<I, O> createChunkProcessor() {
BatchRetryTemplate batchRetryTemplate = createRetryOperations();
FaultTolerantChunkProcessor<I, O> chunkProcessor = new FaultTolerantChunkProcessor<I, O>(getProcessor(),
getWriter(), batchRetryTemplate);
chunkProcessor.setBuffering(!isReaderTransactionalQueue());
chunkProcessor.setProcessorTransactional(processorTransactional);
SkipPolicy writeSkipPolicy = createSkipPolicy();
writeSkipPolicy = getFatalExceptionAwareProxy(writeSkipPolicy);
chunkProcessor.setWriteSkipPolicy(writeSkipPolicy);
chunkProcessor.setProcessSkipPolicy(writeSkipPolicy);
chunkProcessor.setRollbackClassifier(getRollbackClassifier());
chunkProcessor.setKeyGenerator(keyGenerator);
detectStreamInReader();
ArrayList<StepListener> listeners = new ArrayList<StepListener>(getItemListeners());
listeners.addAll(skipListeners);
chunkProcessor.setListeners(listeners);
chunkProcessor.setChunkMonitor(chunkMonitor);
return chunkProcessor;
}
@SuppressWarnings("unchecked")
private void addSpecialExceptions() {
addNonSkippableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
SkipListenerFailedException.class, SkipPolicyFailedException.class, RetryException.class,
JobInterruptedException.class, Error.class);
addNonRetryableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
TransactionException.class, FatalStepExecutionException.class, SkipListenerFailedException.class,
SkipPolicyFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
}
private void detectStreamInReader() {
if (streamIsReader) {
if (!concurrent()) {
chunkMonitor.setItemReader(getReader());
}
else {
logger.warn("Asynchronous TaskExecutor detected with ItemStream reader. This is probably an error, "
+ "and may lead to incorrect restart data being stored.");
}
}
}
/**
* Exception classes that when raised won't crash the job but will result in
* the item which handling caused the exception being skipped. Any exception
* which is marked for "no rollback" is also skippable, but not vice versa.
* Remember to set the {@link #setSkipLimit(int) skip limit} as well.
* <p/>
* Defaults to all no exception.
* Register explicitly set item listeners and auto-register reader, processor and writer if applicable
*/
private void registerSkipListeners() {
// auto-register reader, processor and writer
for (Object itemHandler : new Object[] { getReader(), getWriter(), getProcessor() }) {
if (StepListenerFactoryBean.isListener(itemHandler)) {
StepListener listener = StepListenerFactoryBean.getListener(itemHandler);
if (listener instanceof SkipListener<?, ?>) {
@SuppressWarnings("unchecked")
SkipListener<? super I, ? super O> skipListener = (SkipListener<? super I, ? super O>) listener;
skipListeners.add(skipListener);
}
}
}
}
/**
* Convenience method to get an exception classifier based on the provided transaction attributes.
*
* @param exceptionClasses
* defaults to <code>Exception</code>
* @return an exception classifier: maps to true if an exception should cause rollback
*/
public void setSkippableExceptionClasses(Map<Class<? extends Throwable>, Boolean> exceptionClasses) {
this.skippableExceptionClasses = exceptionClasses;
}
/**
* Exception classes that are candidates for no rollback. The {@link Step}
* can not honour the no rollback hint in all circumstances, but any
* exception on this list is counted as skippable, so even if there has to
* be a rollback, then the step will not fail as long as the skip limit is
* not breached.
* <p/>
* Defaults is empty.
*
* @param noRollbackExceptionClasses
* the exception classes to set
*/
public void setNoRollbackExceptionClasses(Collection<Class<? extends Throwable>> noRollbackExceptionClasses) {
this.noRollbackExceptionClasses = noRollbackExceptionClasses;
}
/**
* @param processorTransactional
*/
public void setProcessorTransactional(boolean processorTransactional) {
this.processorTransactional = processorTransactional;
}
/**
* Convenience method for subclasses to get an exception classifier based on
* the provided transaction attributes.
*
* @return an exception classifier: maps to true if an exception should
* cause rollback
*/
protected Classifier<Throwable, Boolean> getRollbackClassifier() {
private Classifier<Throwable, Boolean> getRollbackClassifier() {
Classifier<Throwable, Boolean> classifier = new BinaryExceptionClassifier(noRollbackExceptionClasses, false);
@@ -317,16 +349,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
}
/**
* Getter for the {@link TransactionAttribute} for subclasses only.
*
* @return the transactionAttribute
*/
@SuppressWarnings("serial")
@Override
protected TransactionAttribute getTransactionAttribute() {
private TransactionAttribute getTransactionAttribute(TransactionAttribute attribute) {
TransactionAttribute attribute = super.getTransactionAttribute();
final Classifier<Throwable, Boolean> classifier = getRollbackClassifier();
return new DefaultTransactionAttribute(attribute) {
@Override
@@ -338,118 +362,33 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
}
@Override
@SuppressWarnings("unchecked")
protected void applyConfiguration(TaskletStep step) {
addNonSkippableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
SkipListenerFailedException.class, SkipPolicyFailedException.class, RetryException.class, JobInterruptedException.class,
Error.class);
addNonRetryableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class, TransactionException.class,
FatalStepExecutionException.class, SkipListenerFailedException.class, SkipPolicyFailedException.class,
RetryException.class, JobInterruptedException.class, Error.class);
super.applyConfiguration(step);
}
/**
* {@inheritDoc}
*/
@Override
protected void registerStreams(TaskletStep step, ItemStream[] streams) {
boolean streamIsReader = false;
ItemReader<? extends T> itemReader = getItemReader();
for (ItemStream stream : streams) {
if (stream instanceof ItemReader<?>) {
streamIsReader = true;
chunkMonitor.registerItemStream(stream);
} else {
step.registerStream(stream);
}
}
TaskExecutor taskExecutor = getTaskExecutor();
// In cases where multiple nested item readers are registered,
// they all want to get the open() and close() callbacks.
if (streamIsReader) {
// double registration is fine
step.registerStream(chunkMonitor);
boolean concurrent = taskExecutor != null && !(taskExecutor instanceof SyncTaskExecutor);
if (!concurrent) {
chunkMonitor.setItemReader(itemReader);
} else {
logger.warn("Asynchronous TaskExecutor detected (" + taskExecutor.getClass()
+ ") with ItemStream reader. This is probably an error, " + "and may lead to incorrect restart data being stored.");
}
}
}
/**
* @return {@link ChunkProvider} configured for fault-tolerance.
*/
@Override
protected SimpleChunkProvider<T> configureChunkProvider() {
SkipPolicy readSkipPolicy = createSkipPolicy();
readSkipPolicy = getFatalExceptionAwareProxy(readSkipPolicy);
FaultTolerantChunkProvider<T> chunkProvider = new FaultTolerantChunkProvider<T>(getItemReader(), getChunkOperations());
chunkProvider.setMaxSkipsOnRead(Math.max(getCommitInterval(), FaultTolerantChunkProvider.DEFAULT_MAX_SKIPS_ON_READ));
chunkProvider.setSkipPolicy(readSkipPolicy);
chunkProvider.setRollbackClassifier(getRollbackClassifier());
return chunkProvider;
}
/**
* @return
*/
protected SkipPolicy createSkipPolicy() {
SkipPolicy skipPolicy = this.skipPolicy;
Map<Class<? extends Throwable>, Boolean> map = new HashMap<Class<? extends Throwable>, Boolean>(skippableExceptionClasses);
Map<Class<? extends Throwable>, Boolean> map = new HashMap<Class<? extends Throwable>, Boolean>(
skippableExceptionClasses);
map.put(ForceRollbackForWriteSkipException.class, true);
LimitCheckingItemSkipPolicy limitCheckingItemSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, map);
LimitCheckingItemSkipPolicy limitCheckingItemSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit , map);
if (skipPolicy == null) {
Assert.state(!(skippableExceptionClasses.isEmpty() && skipLimit > 0),
"If a skip limit is provided then skippable exceptions must also be specified");
skipPolicy = limitCheckingItemSkipPolicy;
} else if (limitCheckingItemSkipPolicy != null) {
}
else if (limitCheckingItemSkipPolicy != null) {
skipPolicy = new CompositeSkipPolicy(new SkipPolicy[] { skipPolicy, limitCheckingItemSkipPolicy });
}
return skipPolicy;
}
/**
* @return {@link ChunkProcessor} configured for fault-tolerance.
*/
@Override
protected SimpleChunkProcessor<T, S> configureChunkProcessor() {
BatchRetryTemplate batchRetryTemplate = configureRetry();
FaultTolerantChunkProcessor<T, S> chunkProcessor = new FaultTolerantChunkProcessor<T, S>(getItemProcessor(), getItemWriter(),
batchRetryTemplate);
chunkProcessor.setBuffering(!isReaderTransactionalQueue());
chunkProcessor.setProcessorTransactional(processorTransactional);
SkipPolicy writeSkipPolicy = createSkipPolicy();
writeSkipPolicy = getFatalExceptionAwareProxy(writeSkipPolicy);
chunkProcessor.setWriteSkipPolicy(writeSkipPolicy);
chunkProcessor.setProcessSkipPolicy(writeSkipPolicy);
chunkProcessor.setRollbackClassifier(getRollbackClassifier());
chunkProcessor.setKeyGenerator(keyGenerator);
chunkProcessor.setChunkMonitor(chunkMonitor);
return chunkProcessor;
}
/**
* @return fully configured retry template for item processing phase.
*/
private BatchRetryTemplate configureRetry() {
private BatchRetryTemplate createRetryOperations() {
RetryPolicy retryPolicy = this.retryPolicy;
SimpleRetryPolicy simpleRetryPolicy = null;
Map<Class<? extends Throwable>, Boolean> map = new HashMap<Class<? extends Throwable>, Boolean>(retryableExceptionClasses);
Map<Class<? extends Throwable>, Boolean> map = new HashMap<Class<? extends Throwable>, Boolean>(
retryableExceptionClasses);
map.put(ForceRollbackForWriteSkipException.class, true);
simpleRetryPolicy = new SimpleRetryPolicy(retryLimit, map);
@@ -457,7 +396,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
Assert.state(!(retryableExceptionClasses.isEmpty() && retryLimit > 0),
"If a retry limit is provided then retryable exceptions must also be specified");
retryPolicy = simpleRetryPolicy;
} else if ((!retryableExceptionClasses.isEmpty() && retryLimit > 0)) {
}
else if ((!retryableExceptionClasses.isEmpty() && retryLimit > 0)) {
CompositeRetryPolicy compositeRetryPolicy = new CompositeRetryPolicy();
compositeRetryPolicy.setPolicies(new RetryPolicy[] { retryPolicy, simpleRetryPolicy });
retryPolicy = compositeRetryPolicy;
@@ -474,28 +414,25 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
// Co-ordinate the retry policy with the exception handler:
RepeatOperations stepOperations = getStepOperations();
if (stepOperations instanceof RepeatTemplate) {
SimpleRetryExceptionHandler exceptionHandler = new SimpleRetryExceptionHandler(retryPolicyWrapper, getExceptionHandler(),
nonRetryableExceptionClasses);
SimpleRetryExceptionHandler exceptionHandler = new SimpleRetryExceptionHandler(retryPolicyWrapper,
getExceptionHandler(), nonRetryableExceptionClasses);
((RepeatTemplate) stepOperations).setExceptionHandler(exceptionHandler);
}
if (retryContextCache == null) {
if (cacheCapacity > 0) {
batchRetryTemplate.setRetryContextCache(new MapRetryContextCache(cacheCapacity));
}
} else {
if (retryContextCache != null) {
batchRetryTemplate.setRetryContextCache(retryContextCache);
}
if (retryListeners != null) {
batchRetryTemplate.setListeners(retryListeners);
batchRetryTemplate.setListeners(retryListeners.toArray(new RetryListener[0]));
}
return batchRetryTemplate;
}
/**
* Wrap the provided {@link #setRetryPolicy(RetryPolicy)} so that it never
* retries explicitly non-retryable exceptions.
* Wrap the provided {@link #setRetryPolicy(RetryPolicy)} so that it never retries explicitly non-retryable
* exceptions.
*/
private RetryPolicy getFatalExceptionAwareProxy(RetryPolicy retryPolicy) {
@@ -505,7 +442,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
map.put(fatal, neverRetryPolicy);
}
SubclassClassifier<Throwable, RetryPolicy> classifier = new SubclassClassifier<Throwable, RetryPolicy>(retryPolicy);
SubclassClassifier<Throwable, RetryPolicy> classifier = new SubclassClassifier<Throwable, RetryPolicy>(
retryPolicy);
classifier.setTypeMap(map);
ExceptionClassifierRetryPolicy retryPolicyWrapper = new ExceptionClassifierRetryPolicy();
@@ -515,11 +453,9 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
}
/**
* Wrap a {@link SkipPolicy} and make it consistent with known fatal
* exceptions.
* Wrap a {@link SkipPolicy} and make it consistent with known fatal exceptions.
*
* @param skipPolicy
* an existing skip policy
* @param skipPolicy an existing skip policy
* @return a skip policy that will not skip fatal exceptions
*/
private SkipPolicy getFatalExceptionAwareProxy(SkipPolicy skipPolicy) {
@@ -564,18 +500,12 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
nonRetryableExceptionClasses = (List<Class<? extends Throwable>>) exceptions;
}
@Override
protected void registerChunkListeners(TaskletStep step, StepListener listener) {
super.registerChunkListeners(step, new TerminateOnExceptionChunkListenerDelegate((ChunkListener) listener));
}
/**
* ChunkListener that wraps exceptions thrown from the ChunkListener in
* {@link FatalStepExecutionException} to force termination of StepExecution
* ChunkListener that wraps exceptions thrown from the ChunkListener in {@link FatalStepExecutionException} to force
* termination of StepExecution
*
* ChunkListeners shoulnd't throw exceptions and expect continued
* processing, they must be handled in the implementation or the step will
* terminate
* ChunkListeners shoulnd't throw exceptions and expect continued processing, they must be handled in the
* implementation or the step will terminate
*
*/
private class TerminateOnExceptionChunkListenerDelegate implements ChunkListener {
@@ -589,7 +519,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
public void beforeChunk() {
try {
chunkListener.beforeChunk();
} catch (Throwable t) {
}
catch (Throwable t) {
throw new FatalStepExecutionException("ChunkListener threw exception, rethrowing as fatal", t);
}
}
@@ -597,7 +528,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
public void afterChunk() {
try {
chunkListener.afterChunk();
} catch (Throwable t) {
}
catch (Throwable t) {
throw new FatalStepExecutionException("ChunkListener threw exception, rethrowing as fatal", t);
}
}

53
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FlowStepBuilder.java Normal file
View File

@@ -0,0 +1,53 @@
/*
* Copyright 2006-2011 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.core.step.builder;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.job.flow.Flow;
import org.springframework.batch.core.job.flow.FlowStep;
/**
* @author Dave Syer
*
*/
public class FlowStepBuilder extends StepBuilderHelper<FlowStepBuilder> {
private Flow flow;
public FlowStepBuilder(StepBuilderHelper<?> parent) {
super(parent);
}
public FlowStepBuilder flow(Flow flow) {
this.flow = flow;
return this;
}
public Step build() {
FlowStep step = new FlowStep();
step.setName(getName());
step.setFlow(flow);
super.enhance(step);
try {
step.afterPropertiesSet();
}
catch (Exception e) {
throw new IllegalStateException(e);
}
return step;
}
}

89
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/JobStepBuilder.java Normal file
View File

@@ -0,0 +1,89 @@
/*
* Copyright 2006-2011 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.core.step.builder;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.batch.core.step.job.JobParametersExtractor;
import org.springframework.batch.core.step.job.JobStep;
/**
* @author Dave Syer
*
*/
public class JobStepBuilder extends StepBuilderHelper<JobStepBuilder> {
private Job job;
private JobLauncher jobLauncher;
private JobParametersExtractor jobParametersExtractor;
public JobStepBuilder(StepBuilderHelper<?> parent) {
super(parent);
}
public JobStepBuilder job(Job job) {
this.job = job;
return this;
}
public JobStepBuilder launcher(JobLauncher jobLauncher) {
this.jobLauncher = jobLauncher;
return this;
}
public JobStepBuilder parametersExtractor(JobParametersExtractor jobParametersExtractor) {
this.jobParametersExtractor = jobParametersExtractor;
return this;
}
public Step build() {
JobStep step = new JobStep();
step.setName(getName());
super.enhance(step);
if (job != null) {
step.setJob(job);
}
if (jobParametersExtractor != null) {
step.setJobParametersExtractor(jobParametersExtractor);
}
if (jobLauncher == null) {
SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
jobLauncher.setJobRepository(getJobRepository());
try {
jobLauncher.afterPropertiesSet();
}
catch (Exception e) {
throw new StepBuilderException(e);
}
this.jobLauncher = jobLauncher;
}
step.setJobLauncher(jobLauncher);
try {
step.afterPropertiesSet();
}
catch (Exception e) {
throw new IllegalStateException(e);
}
return step;
}
}

154
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/PartitionStepBuilder.java Normal file
View File

@@ -0,0 +1,154 @@
/*
* Copyright 2006-2011 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.core.step.builder;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.partition.PartitionHandler;
import org.springframework.batch.core.partition.StepExecutionSplitter;
import org.springframework.batch.core.partition.support.PartitionStep;
import org.springframework.batch.core.partition.support.Partitioner;
import org.springframework.batch.core.partition.support.SimpleStepExecutionSplitter;
import org.springframework.batch.core.partition.support.StepExecutionAggregator;
import org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
/**
* @author Dave Syer
*
*/
public class PartitionStepBuilder extends StepBuilderHelper<PartitionStepBuilder> {
private TaskExecutor taskExecutor;
private Partitioner partitioner;
private static final int DEFAULT_GRID_SIZE = 6;
private Step step;
private PartitionHandler partitionHandler;
private int gridSize = DEFAULT_GRID_SIZE;
private StepExecutionSplitter splitter;
private StepExecutionAggregator aggregator;
private String stepName;
public PartitionStepBuilder(StepBuilderHelper<?> parent) {
super(parent);
}
public PartitionStepBuilder partitioner(String slaveStepName, Partitioner partitioner) {
this.stepName = slaveStepName;
this.partitioner = partitioner;
return this;
}
public PartitionStepBuilder step(Step step) {
this.step = step;
return this;
}
public PartitionStepBuilder taskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
return this;
}
public PartitionStepBuilder partitionHandler(PartitionHandler partitionHandler) {
this.partitionHandler = partitionHandler;
return this;
}
public PartitionStepBuilder gridSize(int gridSize) {
this.gridSize = gridSize;
return this;
}
public Step build() {
PartitionStep step = new PartitionStep();
step.setName(getName());
super.enhance(step);
if (partitionHandler != null) {
step.setPartitionHandler(partitionHandler);
}
else {
TaskExecutorPartitionHandler partitionHandler = new TaskExecutorPartitionHandler();
partitionHandler.setStep(this.step);
if (taskExecutor == null) {
taskExecutor = new SyncTaskExecutor();
}
partitionHandler.setGridSize(gridSize);
partitionHandler.setTaskExecutor(taskExecutor);
step.setPartitionHandler(partitionHandler);
}
if (splitter!=null) {
step.setStepExecutionSplitter(splitter);
} else {
boolean allowStartIfComplete = isAllowStartIfComplete();
String name = stepName;
if (this.step != null) {
try {
allowStartIfComplete = this.step.isAllowStartIfComplete();
name = this.step.getName();
}
catch (Exception e) {
logger.info("Ignored exception from step asking for name and allowStartIfComplete flag. "
+ "Using default from enclosing PartitionStep (" + name + "," + allowStartIfComplete + ").");
}
}
SimpleStepExecutionSplitter splitter = new SimpleStepExecutionSplitter();
splitter.setPartitioner(partitioner);
splitter.setJobRepository(getJobRepository());
splitter.setAllowStartIfComplete(allowStartIfComplete);
splitter.setStepName(name);
this.splitter = splitter;
step.setStepExecutionSplitter(splitter);
}
if (aggregator!=null) {
step.setStepExecutionAggregator(aggregator);
}
try {
step.afterPropertiesSet();
}
catch (Exception e) {
throw new IllegalStateException(e);
}
return step;
}
public PartitionStepBuilder splitter(StepExecutionSplitter splitter) {
this.splitter = splitter;
return this;
}
public PartitionStepBuilder aggregator(StepExecutionAggregator aggregator) {
this.aggregator = aggregator;
return this;
}
}

225
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java Normal file
View File

@@ -0,0 +1,225 @@
/*
* Copyright 2006-2011 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.core.step.builder;
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.Set;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.ItemProcessListener;
import org.springframework.batch.core.ItemReadListener;
import org.springframework.batch.core.ItemWriteListener;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.listener.StepListenerFactoryBean;
import org.springframework.batch.core.step.item.ChunkOrientedTasklet;
import org.springframework.batch.core.step.item.SimpleChunkProcessor;
import org.springframework.batch.core.step.item.SimpleChunkProvider;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.CompletionPolicy;
import org.springframework.batch.repeat.RepeatOperations;
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.util.Assert;
/**
* @author Dave Syer
*
*/
public class SimpleStepBuilder<I, O> extends AbstractTaskletStepBuilder<SimpleStepBuilder<I, O>> {
private static final int DEFAULT_COMMIT_INTERVAL = 1;
private ItemReader<? extends I> reader;
private ItemWriter<? super O> writer;
private ItemProcessor<? super I, ? extends O> processor;
private int chunkSize = 0;
private RepeatOperations chunkOperations;
private CompletionPolicy completionPolicy;
private Set<StepListener> itemListeners = new LinkedHashSet<StepListener>();
private boolean readerTransactionalQueue = false;
public SimpleStepBuilder(StepBuilderHelper<?> parent) {
super(parent);
}
@Override
public TaskletStep build() {
registerAsStreamsAndListeners(reader, processor, writer);
return super.build();
}
@Override
protected Tasklet createTasklet() {
Assert.state(reader != null, "ItemReader must be provided");
Assert.state(processor != null || writer != null, "ItemWriter or ItemProcessor must be provided");
RepeatOperations repeatOperations = createChunkOperations();
SimpleChunkProvider<I> chunkProvider = new SimpleChunkProvider<I>(reader, repeatOperations);
SimpleChunkProcessor<I, O> chunkProcessor = new SimpleChunkProcessor<I, O>(processor, writer);
chunkProvider.setListeners(new ArrayList<StepListener>(itemListeners));
chunkProcessor.setListeners(new ArrayList<StepListener>(itemListeners));
ChunkOrientedTasklet<I> tasklet = new ChunkOrientedTasklet<I>(chunkProvider, chunkProcessor);
tasklet.setBuffering(!readerTransactionalQueue);
return tasklet;
}
public SimpleStepBuilder<I, O> readerIsTransactionalQueue() {
this.readerTransactionalQueue = true;
return this;
}
public SimpleStepBuilder<I, O> listener(ItemReadListener<? super I> listener) {
itemListeners.add(listener);
return this;
}
public SimpleStepBuilder<I, O> listener(ItemWriteListener<? super O> listener) {
itemListeners.add(listener);
return this;
}
public SimpleStepBuilder<I, O> listener(ItemProcessListener<? super I, ? super O> listener) {
itemListeners.add(listener);
return this;
}
public SimpleStepBuilder<I, O> chunkOperations(RepeatOperations repeatTemplate) {
this.chunkOperations = repeatTemplate;
return this;
}
public SimpleStepBuilder<I, O> completionPolicy(CompletionPolicy completionPolicy) {
Assert.state(chunkSize == 0 || completionPolicy == null,
"You must specify either a chunkCompletionPolicy or a commitInterval but not both.");
this.completionPolicy = completionPolicy;
return this;
}
public SimpleStepBuilder<I, O> chunk(int chunkSize) {
Assert.state(completionPolicy == null || chunkSize == 0,
"You must specify either a chunkCompletionPolicy or a commitInterval but not both.");
this.chunkSize = chunkSize;
return this;
}
public SimpleStepBuilder<I, O> chunk(CompletionPolicy completionPolicy) {
this.completionPolicy = completionPolicy;
return this;
}
public SimpleStepBuilder<I, O> reader(ItemReader<? extends I> reader) {
this.reader = reader;
return this;
}
public SimpleStepBuilder<I, O> writer(ItemWriter<? super O> writer) {
this.writer = writer;
return this;
}
public SimpleStepBuilder<I, O> processor(ItemProcessor<? super I, ? extends O> processor) {
this.processor = processor;
return this;
}
protected RepeatOperations createChunkOperations() {
RepeatOperations repeatOperations = chunkOperations;
if (repeatOperations == null) {
RepeatTemplate repeatTemplate = new RepeatTemplate();
repeatTemplate.setCompletionPolicy(getChunkCompletionPolicy());
repeatOperations = repeatTemplate;
}
return repeatOperations;
}
protected ItemReader<? extends I> getReader() {
return reader;
}
protected ItemWriter<? super O> getWriter() {
return writer;
}
protected ItemProcessor<? super I, ? extends O> getProcessor() {
return processor;
}
protected int getChunkSize() {
return chunkSize;
}
protected boolean isReaderTransactionalQueue() {
return readerTransactionalQueue;
}
protected Set<StepListener> getItemListeners() {
return itemListeners;
}
/**
* @return a {@link CompletionPolicy} consistent with the chunk size and injected policy (if present).
*/
private CompletionPolicy getChunkCompletionPolicy() {
Assert.state(!(completionPolicy != null && chunkSize > 0),
"You must specify either a chunkCompletionPolicy or a commitInterval but not both.");
Assert.state(chunkSize >= 0, "The commitInterval must be positive or zero (for default value).");
if (completionPolicy != null) {
return completionPolicy;
}
if (chunkSize == 0) {
logger.info("Setting commit interval to default value (" + DEFAULT_COMMIT_INTERVAL + ")");
chunkSize = DEFAULT_COMMIT_INTERVAL;
}
return new SimpleCompletionPolicy(chunkSize);
}
private void registerAsStreamsAndListeners(ItemReader<? extends I> itemReader,
ItemProcessor<? super I, ? extends O> itemProcessor, ItemWriter<? super O> itemWriter) {
for (Object itemHandler : new Object[] { itemReader, itemWriter, itemProcessor }) {
if (itemHandler instanceof ItemStream) {
stream((ItemStream) itemHandler);
}
if (StepListenerFactoryBean.isListener(itemHandler)) {
StepListener listener = StepListenerFactoryBean.getListener(itemHandler);
if (listener instanceof StepExecutionListener) {
listener((StepExecutionListener) listener);
}
if (listener instanceof ChunkListener) {
listener((ChunkListener) listener);
}
if (listener instanceof ItemReadListener<?> || listener instanceof ItemProcessListener<?, ?>
|| listener instanceof ItemWriteListener<?>) {
itemListeners.add(listener);
}
}
}
}
}

64
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilder.java Normal file
View File

@@ -0,0 +1,64 @@
/*
* Copyright 2006-2011 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.core.step.builder;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.job.flow.Flow;
import org.springframework.batch.core.partition.support.Partitioner;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.CompletionPolicy;
/**
* @author Dave Syer
*
*/
public class StepBuilder extends StepBuilderHelper<StepBuilder> {
public StepBuilder(String name) {
super(name);
}
public TaskletStepBuilder tasklet(Tasklet tasklet) {
return new TaskletStepBuilder(this).tasklet(tasklet);
}
public <I, O> SimpleStepBuilder<I, O> chunk(int chunkSize) {
return new SimpleStepBuilder<I, O>(this).chunk(chunkSize);
}
public <I, O> SimpleStepBuilder<I, O> chunk(CompletionPolicy completionPolicy) {
return new SimpleStepBuilder<I, O>(this).completionPolicy(completionPolicy);
}
public PartitionStepBuilder partitioner(String stepName, Partitioner partitioner) {
return new PartitionStepBuilder(this).partitioner(stepName, partitioner);
}
public PartitionStepBuilder partitioner(Step step) {
return new PartitionStepBuilder(this).step(step);
}
public JobStepBuilder job(Job job) {
return new JobStepBuilder(this).job(job);
}
public FlowStepBuilder flow(Flow flow) {
return new FlowStepBuilder(this).flow(flow);
}
}

24
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderException.java Normal file
View File

@@ -0,0 +1,24 @@
/*
* Copyright 2006-2011 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.core.step.builder;
public class StepBuilderException extends RuntimeException {
public StepBuilderException(Exception e) {
super(e);
}
}

186
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderHelper.java Normal file
View File

@@ -0,0 +1,186 @@
/*
* Copyright 2006-2011 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.core.step.builder;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.AbstractStep;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.transaction.PlatformTransactionManager;
/**
* @author Dave Syer
*
*/
public abstract class StepBuilderHelper<B extends StepBuilderHelper<B>> {
protected final Log logger = LogFactory.getLog(getClass());
private final CommonStepProperties properties;
public StepBuilderHelper(String name) {
this.properties = new CommonStepProperties();
properties.name = name;
}
protected StepBuilderHelper(StepBuilderHelper<?> parent) {
this.properties = parent.properties;
}
public StepBuilderHelper<B> repository(JobRepository jobRepository) {
properties.jobRepository = jobRepository;
return this;
}
public StepBuilderHelper<B> transactionManager(PlatformTransactionManager transactionManager) {
properties.transactionManager = transactionManager;
return this;
}
public StepBuilderHelper<B> startLimit(int startLimit) {
properties.startLimit = startLimit;
return this;
}
public StepBuilderHelper<B> listener(StepExecutionListener listener) {
properties.addStepExecutionListener(listener);
return this;
}
public StepBuilderHelper<B> allowStartIfComplete(boolean allowStartIfComplete) {
properties.allowStartIfComplete = allowStartIfComplete;
return this;
}
protected String getName() {
return properties.name;
}
protected JobRepository getJobRepository() {
return properties.jobRepository;
}
protected PlatformTransactionManager getTransactionManager() {
return properties.transactionManager;
}
protected boolean isAllowStartIfComplete() {
return properties.allowStartIfComplete != null ? properties.allowStartIfComplete : false;
}
protected void enhance(Step target) {
if (target instanceof AbstractStep) {
AbstractStep step = (AbstractStep) target;
step.setJobRepository(properties.getJobRepository());
Boolean allowStartIfComplete = properties.allowStartIfComplete;
if (allowStartIfComplete != null) {
step.setAllowStartIfComplete(allowStartIfComplete);
}
step.setStartLimit(properties.startLimit);
List<StepExecutionListener> listeners = properties.stepExecutionListeners;
if (!listeners.isEmpty()) {
step.setStepExecutionListeners(listeners.toArray(new StepExecutionListener[0]));
}
}
if (target instanceof TaskletStep) {
TaskletStep step = (TaskletStep) target;
step.setTransactionManager(properties.transactionManager);
}
}
public static class CommonStepProperties {
private List<StepExecutionListener> stepExecutionListeners = new ArrayList<StepExecutionListener>();
private int startLimit = Integer.MAX_VALUE;
private Boolean allowStartIfComplete;
private JobRepository jobRepository;
private PlatformTransactionManager transactionManager;
public JobRepository getJobRepository() {
return jobRepository;
}
public void setJobRepository(JobRepository jobRepository) {
this.jobRepository = jobRepository;
}
public PlatformTransactionManager getTransactionManager() {
return transactionManager;
}
public void setTransactionManager(PlatformTransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public List<StepExecutionListener> getStepExecutionListeners() {
return stepExecutionListeners;
}
public void addStepExecutionListeners(List<StepExecutionListener> stepExecutionListeners) {
this.stepExecutionListeners.addAll(stepExecutionListeners);
}
public void addStepExecutionListener(StepExecutionListener stepExecutionListener) {
this.stepExecutionListeners.add(stepExecutionListener);
}
public Integer getStartLimit() {
return startLimit;
}
public void setStartLimit(Integer startLimit) {
this.startLimit = startLimit;
}
public Boolean getAllowStartIfComplete() {
return allowStartIfComplete;
}
public void setAllowStartIfComplete(Boolean allowStartIfComplete) {
this.allowStartIfComplete = allowStartIfComplete;
}
private String name;
}
}

42
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/TaskletStepBuilder.java Normal file
View File

@@ -0,0 +1,42 @@
/*
* Copyright 2006-2011 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.core.step.builder;
import org.springframework.batch.core.step.tasklet.Tasklet;
/**
* @author Dave Syer
*
*/
public class TaskletStepBuilder extends AbstractTaskletStepBuilder<TaskletStepBuilder> {
private Tasklet tasklet;
public TaskletStepBuilder(StepBuilderHelper<?> parent) {
super(parent);
}
public TaskletStepBuilder tasklet(Tasklet tasklet) {
this.tasklet = tasklet;
return this;
}
@Override
protected Tasklet createTasklet() {
return tasklet;
}
}

5
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java → spring-batch-core/src/main/java/org/springframework/batch/core/step/factory/BatchListenerFactoryHelper.java
View File

@@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.step.item;
package org.springframework.batch.core.step.factory;
import java.util.ArrayList;
import java.util.List;
@@ -30,8 +30,7 @@ abstract class BatchListenerFactoryHelper {
public static <T> List<T> getListeners(StepListener[] listeners, Class<? super T> cls) {
List<T> list = new ArrayList<T>();
for (int i = 0; i < listeners.length; i++) {
StepListener stepListener = listeners[i];
for (StepListener stepListener : listeners) {
if (cls.isAssignableFrom(stepListener.getClass())) {
@SuppressWarnings("unchecked")
T listener = (T) stepListener;

280
spring-batch-core/src/main/java/org/springframework/batch/core/step/factory/FaultTolerantStepFactoryBean.java Executable file
View File

@@ -0,0 +1,280 @@
/*
* 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.core.step.factory;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.step.builder.FaultTolerantStepBuilder;
import org.springframework.batch.core.step.builder.SimpleStepBuilder;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.core.step.item.KeyGenerator;
import org.springframework.batch.core.step.skip.SkipPolicy;
import org.springframework.retry.RetryListener;
import org.springframework.retry.RetryPolicy;
import org.springframework.retry.backoff.BackOffPolicy;
import org.springframework.retry.policy.MapRetryContextCache;
import org.springframework.retry.policy.RetryContextCache;
/**
* Factory bean for step that provides options for configuring skip behaviour. User can set {@link #setSkipLimit(int)}
* to set how many exceptions of {@link #setSkippableExceptionClasses(Collection)} types are tolerated.
* {@link #setFatalExceptionClasses(Collection)} will cause immediate termination of job - they are treated as higher
* priority than {@link #setSkippableExceptionClasses(Collection)}, so the two lists don't need to be exclusive.
*
* Skippable exceptions on write will by default cause transaction rollback - to avoid rollback for specific exception
* class include it in the transaction attribute as "no rollback for".
*
* @see SimpleStepFactoryBean
*
* @author Dave Syer
* @author Robert Kasanicky
* @author Morten Andersen-Gott
*
*/
public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S> {
private Map<Class<? extends Throwable>, Boolean> skippableExceptionClasses = new HashMap<Class<? extends Throwable>, Boolean>();
private Collection<Class<? extends Throwable>> noRollbackExceptionClasses = new HashSet<Class<? extends Throwable>>();
private Map<Class<? extends Throwable>, Boolean> retryableExceptionClasses = new HashMap<Class<? extends Throwable>, Boolean>();
private int cacheCapacity = 0;
private int retryLimit = 0;
private int skipLimit = 0;
private SkipPolicy skipPolicy;
private BackOffPolicy backOffPolicy;
private RetryListener[] retryListeners;
private RetryPolicy retryPolicy;
private RetryContextCache retryContextCache;
private KeyGenerator keyGenerator;
private boolean processorTransactional = true;
/**
* The {@link KeyGenerator} to use to identify failed items across rollback. Not used in the case of the
* {@link #setIsReaderTransactionalQueue(boolean) transactional queue flag} being false (the default).
*
* @param keyGenerator the {@link KeyGenerator} to set
*/
public void setKeyGenerator(KeyGenerator keyGenerator) {
this.keyGenerator = keyGenerator;
}
/**
* Setter for the retry policy. If this is specified the other retry properties are ignored (retryLimit,
* backOffPolicy, retryableExceptionClasses).
*
* @param retryPolicy a stateless {@link RetryPolicy}
*/
public void setRetryPolicy(RetryPolicy retryPolicy) {
this.retryPolicy = retryPolicy;
}
/**
* Public setter for the retry limit. Each item can be retried up to this limit. Note this limit includes the
* initial attempt to process the item, therefore <code>retryLimit == 1</code> by default.
*
* @param retryLimit the retry limit to set, must be greater or equal to 1.
*/
public void setRetryLimit(int retryLimit) {
this.retryLimit = retryLimit;
}
/**
* Public setter for the capacity of the cache in the retry policy. If more items than this fail without being
* skipped or recovered an exception will be thrown. This is to guard against inadvertent infinite loops generated
* by item identity problems.<br/>
*
* The default value should be high enough and more for most purposes. To breach the limit in a single-threaded step
* typically you have to have this many failures in a single transaction. Defaults to the value in the
* {@link MapRetryContextCache}.<br/>
*
* This property is ignored if the {@link #setRetryContextCache(RetryContextCache)} is set directly.
*
* @param cacheCapacity the cache capacity to set (greater than 0 else ignored)
*/
public void setCacheCapacity(int cacheCapacity) {
this.cacheCapacity = cacheCapacity;
}
/**
* Override the default retry context cache for retry of chunk processing. If this property is set then
* {@link #setCacheCapacity(int)} is ignored.
*
* @param retryContextCache the {@link RetryContextCache} to set
*/
public void setRetryContextCache(RetryContextCache retryContextCache) {
this.retryContextCache = retryContextCache;
}
/**
* Public setter for the retryable exceptions classifier map (from throwable class to boolean, true is retryable).
*
* @param retryableExceptionClasses the retryableExceptionClasses to set
*/
public void setRetryableExceptionClasses(Map<Class<? extends Throwable>, Boolean> 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;
}
/**
* A limit that determines skip policy. If this value is positive then an exception in chunk processing will cause
* the item to be skipped and no exception propagated until the limit is reached. If it is zero then all exceptions
* will be propagated from the chunk and cause the step to abort.
*
* @param skipLimit the value to set. Default is 0 (never skip).
*/
public void setSkipLimit(int skipLimit) {
this.skipLimit = skipLimit;
}
/**
* A {@link SkipPolicy} that determines the outcome of an exception when processing an item. Overrides the
* {@link #setSkipLimit(int) skipLimit}. The {@link #setSkippableExceptionClasses(Map) skippableExceptionClasses}
* are also ignored if this is set.
*
* @param skipPolicy the {@link SkipPolicy} to set
*/
public void setSkipPolicy(SkipPolicy skipPolicy) {
this.skipPolicy = skipPolicy;
}
/**
* Exception classes that when raised won't crash the job but will result in the item which handling caused the
* exception being skipped. Any exception which is marked for "no rollback" is also skippable, but not vice versa.
* Remember to set the {@link #setSkipLimit(int) skip limit} as well.
* <p/>
* Defaults to all no exception.
*
* @param exceptionClasses defaults to <code>Exception</code>
*/
public void setSkippableExceptionClasses(Map<Class<? extends Throwable>, Boolean> exceptionClasses) {
this.skippableExceptionClasses = exceptionClasses;
}
/**
* Exception classes that are candidates for no rollback. The {@link Step} can not honour the no rollback hint in
* all circumstances, but any exception on this list is counted as skippable, so even if there has to be a rollback,
* then the step will not fail as long as the skip limit is not breached.
* <p/>
* Defaults is empty.
*
* @param noRollbackExceptionClasses the exception classes to set
*/
public void setNoRollbackExceptionClasses(Collection<Class<? extends Throwable>> noRollbackExceptionClasses) {
this.noRollbackExceptionClasses = noRollbackExceptionClasses;
}
/**
* @param processorTransactional
*/
public void setProcessorTransactional(boolean processorTransactional) {
this.processorTransactional = processorTransactional;
}
protected SimpleStepBuilder<T, S> createBuilder(String name) {
return new FaultTolerantStepBuilder<T, S>(new StepBuilder(name));
}
@Override
protected void applyConfiguration(SimpleStepBuilder<T, S> builder) {
FaultTolerantStepBuilder<T, S> faultTolerantBuilder = (FaultTolerantStepBuilder<T, S>) builder;
if (retryContextCache == null && cacheCapacity > 0) {
retryContextCache = new MapRetryContextCache(cacheCapacity);
}
faultTolerantBuilder.retryContextCache(retryContextCache);
for (SkipListener<T, S> listener : BatchListenerFactoryHelper.<SkipListener<T, S>> getListeners(getListeners(),
SkipListener.class)) {
faultTolerantBuilder.listener(listener);
}
if (retryListeners != null) {
for (RetryListener listener : retryListeners) {
faultTolerantBuilder.listener(listener);
}
}
faultTolerantBuilder.skipPolicy(skipPolicy);
faultTolerantBuilder.skipLimit(skipLimit);
for (Class<? extends Throwable> type : skippableExceptionClasses.keySet()) {
if (skippableExceptionClasses.get(type)) {
faultTolerantBuilder.skip(type);
}
else {
faultTolerantBuilder.noSkip(type);
}
}
if (!processorTransactional) {
faultTolerantBuilder.processorNonTransactional();
}
faultTolerantBuilder.retryContextCache(retryContextCache);
faultTolerantBuilder.keyGenerator(keyGenerator);
faultTolerantBuilder.retryPolicy(retryPolicy);
faultTolerantBuilder.retryLimit(retryLimit);
faultTolerantBuilder.backOffPolicy(backOffPolicy);
for (Class<? extends Throwable> type : retryableExceptionClasses.keySet()) {
if (retryableExceptionClasses.get(type)) {
faultTolerantBuilder.retry(type);
}
else {
faultTolerantBuilder.noRetry(type);
}
}
for (Class<? extends Throwable> type : noRollbackExceptionClasses) {
faultTolerantBuilder.noRollback(type);
}
super.applyConfiguration(builder);
}
}

326
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java → spring-batch-core/src/main/java/org/springframework/batch/core/step/factory/SimpleStepFactoryBean.java
View File

@@ -13,10 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.step.item;
import java.util.Arrays;
import java.util.List;
package org.springframework.batch.core.step.factory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -24,12 +21,12 @@ import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.ItemProcessListener;
import org.springframework.batch.core.ItemReadListener;
import org.springframework.batch.core.ItemWriteListener;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.listener.StepListenerFactoryBean;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.SimpleStepBuilder;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
@@ -40,7 +37,6 @@ import org.springframework.batch.repeat.RepeatOperations;
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;
import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.FactoryBean;
@@ -50,14 +46,13 @@ import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
import org.springframework.transaction.interceptor.TransactionAttribute;
import org.springframework.util.Assert;
/**
* Most common configuration options for simple steps should be found here. Use
* this factory bean instead of creating a {@link Step} implementation manually.
* Most common configuration options for simple steps should be found here. Use this factory bean instead of creating a
* {@link Step} implementation manually.
*
* This factory does not support configuration of fault-tolerant behavior, use
* appropriate subclass of this factory bean to configure skip or retry.
* This factory does not support configuration of fault-tolerant behavior, use appropriate subclass of this factory bean
* to configure skip or retry.
*
* @see FaultTolerantStepFactoryBean
*
@@ -67,8 +62,6 @@ import org.springframework.util.Assert;
*/
public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
private static final int DEFAULT_COMMIT_INTERVAL = 1;
private String name;
private int startLimit = Integer.MAX_VALUE;
@@ -123,9 +116,8 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
}
/**
* Flag to signal that the reader is transactional (usually a JMS consumer)
* so that items are re-presented after a rollback. The default is false and
* readers are assumed to be forward-only.
* Flag to signal that the reader is transactional (usually a JMS consumer) so that items are re-presented after a
* rollback. The default is false and readers are assumed to be forward-only.
*
* @param isReaderTransactionalQueue the value of the flag
*/
@@ -142,8 +134,7 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
}
/**
* Set the bean name property, which will become the name of the
* {@link Step} when it is created.
* Set the bean name property, which will become the name of the {@link Step} when it is created.
*
* @see org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang.String)
*/
@@ -162,8 +153,7 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
/**
* The timeout for an individual transaction in the step.
*
* @param transactionTimeout the transaction timeout to set, defaults to
* infinite
* @param transactionTimeout the transaction timeout to set, defaults to infinite
*/
public void setTransactionTimeout(int transactionTimeout) {
this.transactionTimeout = transactionTimeout;
@@ -193,8 +183,8 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
}
/**
* Public setter for the flag to indicate that the step should be replayed
* on a restart, even if successful the first time.
* Public setter for the flag to indicate that the step should be replayed on a restart, even if successful the
* first time.
*
* @param allowStartIfComplete the shouldAllowStartIfComplete to set
*/
@@ -224,9 +214,8 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
}
/**
* The streams to inject into the {@link Step}. Any instance of
* {@link ItemStream} can be used, and will then receive callbacks at the
* appropriate stage in the step.
* The streams to inject into the {@link Step}. Any instance of {@link ItemStream} can be used, and will then
* receive callbacks at the appropriate stage in the step.
*
* @param streams an array of listeners
*/
@@ -235,9 +224,8 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
}
/**
* The listeners to inject into the {@link Step}. Any instance of
* {@link StepListener} can be used, and will then receive callbacks at the
* appropriate stage in the step.
* The listeners to inject into the {@link Step}. Any instance of {@link StepListener} can be used, and will then
* receive callbacks at the appropriate stage in the step.
*
* @param listeners an array of listeners
*/
@@ -309,9 +297,8 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
return new DefaultTransactionAttribute(attribute) {
/**
* Ignore the default behaviour and rollback on all exceptions that
* bubble up to the tasklet level. The tasklet has to deal with the
* rollback rules internally.
* Ignore the default behaviour and rollback on all exceptions that bubble up to the tasklet level. The
* tasklet has to deal with the rollback rules internally.
*/
@Override
public boolean rollbackOn(Throwable ex) {
@@ -328,20 +315,23 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
* @see FactoryBean#getObject()
*/
public final Object getObject() throws Exception {
TaskletStep step = new TaskletStep(getName());
applyConfiguration(step);
step.afterPropertiesSet();
SimpleStepBuilder<T, S> builder = createBuilder(getName());
applyConfiguration(builder);
TaskletStep step = builder.build();
return step;
}
protected SimpleStepBuilder<T, S> createBuilder(String name) {
return new SimpleStepBuilder<T, S>(new StepBuilder(name));
}
public Class<TaskletStep> getObjectType() {
return TaskletStep.class;
}
/**
* Returns true by default, but in most cases a {@link Step} should not be
* treated as thread safe. Clients are recommended to create a new step for
* each job execution.
* Returns true by default, but in most cases a {@link Step} should not be treated as thread safe. Clients are
* recommended to create a new step for each job execution.
*
* @see org.springframework.beans.factory.FactoryBean#isSingleton()
*/
@@ -358,29 +348,18 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
}
/**
* Set the commit interval. Either set this or the chunkCompletionPolicy but
* not both.
* Set the commit interval. Either set this or the chunkCompletionPolicy but not both.
*
* @param commitInterval 1 by default
*/
public void setCommitInterval(int commitInterval) {
this.commitInterval = commitInterval;
}
/**
* Accessor for commit interval if needed in sub classes.
*
* @return the commit interval
*/
protected int getCommitInterval() {
return commitInterval;
}
/**
* Public setter for the {@link CompletionPolicy} applying to the chunk
* level. A transaction will be committed when this policy decides to
* complete. Defaults to a {@link SimpleCompletionPolicy} with chunk size
* equal to the commitInterval property.
* Public setter for the {@link CompletionPolicy} applying to the chunk level. A transaction will be committed when
* this policy decides to complete. Defaults to a {@link SimpleCompletionPolicy} with chunk size equal to the
* commitInterval property.
*
* @param chunkCompletionPolicy the chunkCompletionPolicy to set
*/
@@ -389,8 +368,7 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
}
/**
* Protected getter for the step operations to make them available in
* subclasses.
* Protected getter for the step operations to make them available in subclasses.
* @return the step operations
*/
protected RepeatOperations getStepOperations() {
@@ -414,8 +392,7 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
}
/**
* Protected getter for the chunk operations to make them available in
* subclasses.
* Protected getter for the chunk operations to make them available in subclasses.
* @return the step operations
*/
protected RepeatOperations getChunkOperations() {
@@ -439,8 +416,8 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
}
/**
* Public setter for the {@link TaskExecutor}. If this is set, then it will
* be used to execute the chunk processing inside the {@link Step}.
* 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
*/
@@ -457,202 +434,57 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
}
/**
* Public setter for the throttle limit. This limits the number of tasks
* queued for concurrent processing to prevent thread pools from being
* overwhelmed. Defaults to
* {@link TaskExecutorRepeatTemplate#DEFAULT_THROTTLE_LIMIT}.
* Public setter for the throttle limit. This limits the number of tasks queued for concurrent processing to prevent
* thread pools from being overwhelmed. Defaults to {@link TaskExecutorRepeatTemplate#DEFAULT_THROTTLE_LIMIT}.
* @param throttleLimit the throttle limit to set.
*/
public void setThrottleLimit(int throttleLimit) {
this.throttleLimit = throttleLimit;
}
/**
* @param step
*
*/
protected void applyConfiguration(TaskletStep step) {
protected void applyConfiguration(SimpleStepBuilder<T, S> builder) {
Assert.state(getItemReader()!=null, "ItemReader must be provided");
Assert.state(getItemWriter()!=null || getItemProcessor()!=null, "ItemWriter or ItemProcessor must be provided");
Assert.state(transactionManager!=null, "TransactionManager must be provided");
step.setTransactionManager(transactionManager);
step.setTransactionAttribute(getTransactionAttribute());
step.setJobRepository(jobRepository);
step.setStartLimit(startLimit);
step.setAllowStartIfComplete(allowStartIfComplete);
registerStreams(step, streams);
if (chunkOperations == null) {
RepeatTemplate repeatTemplate = new RepeatTemplate();
repeatTemplate.setCompletionPolicy(getChunkCompletionPolicy());
chunkOperations = repeatTemplate;
builder.reader(itemReader);
builder.processor(itemProcessor);
builder.writer(itemWriter);
for (StepExecutionListener listener : BatchListenerFactoryHelper.<StepExecutionListener> getListeners(
listeners, StepExecutionListener.class)) {
builder.listener(listener);
}
for (ChunkListener listener : BatchListenerFactoryHelper.<ChunkListener> getListeners(listeners,
ChunkListener.class)) {
builder.listener(listener);
}
for (ItemReadListener<T> listener : BatchListenerFactoryHelper.<ItemReadListener<T>> getListeners(listeners,
ItemReadListener.class)) {
builder.listener(listener);
}
for (ItemWriteListener<S> listener : BatchListenerFactoryHelper.<ItemWriteListener<S>> getListeners(listeners,
ItemWriteListener.class)) {
builder.listener(listener);
}
for (ItemProcessListener<T, S> listener : BatchListenerFactoryHelper.<ItemProcessListener<T, S>> getListeners(
listeners, ItemProcessListener.class)) {
builder.listener(listener);
}
builder.transactionManager(transactionManager);
builder.transactionAttribute(getTransactionAttribute());
builder.repository(jobRepository);
builder.startLimit(startLimit);
builder.allowStartIfComplete(allowStartIfComplete);
builder.chunk(commitInterval);
builder.completionPolicy(chunkCompletionPolicy);
builder.chunkOperations(chunkOperations);
builder.stepOperations(stepOperations);
builder.taskExecutor(taskExecutor);
builder.throttleLimit(throttleLimit);
builder.exceptionHandler(exceptionHandler);
if (isReaderTransactionalQueue) {
builder.readerIsTransactionalQueue();
}
for (ItemStream stream : streams) {
builder.stream(stream);
}
if (stepOperations == null) {
stepOperations = new RepeatTemplate();
if (taskExecutor != null) {
TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate();
repeatTemplate.setTaskExecutor(taskExecutor);
repeatTemplate.setThrottleLimit(throttleLimit);
stepOperations = repeatTemplate;
}
((RepeatTemplate) stepOperations).setExceptionHandler(exceptionHandler);
}
step.setStepOperations(stepOperations);
SimpleChunkProvider<T> chunkProvider = configureChunkProvider();
SimpleChunkProcessor<T, S> chunkProcessor = configureChunkProcessor();
registerItemListeners(chunkProvider, chunkProcessor);
registerStepListeners(step, chunkOperations);
registerStreams(step, itemReader, itemProcessor, itemWriter);
ChunkOrientedTasklet<T> tasklet = new ChunkOrientedTasklet<T>(chunkProvider, chunkProcessor);
tasklet.setBuffering(!isReaderTransactionalQueue());
step.setTasklet(tasklet);
}
/**
* Register the streams with the step.
* @param step the {@link TaskletStep}
* @param streams the streams to register
*/
protected void registerStreams(TaskletStep step, ItemStream[] streams) {
step.setStreams(streams);
}
/**
* Extension point for creating appropriate {@link ChunkProvider}. Return
* value must subclass {@link SimpleChunkProvider} due to listener
* registration.
*/
protected SimpleChunkProvider<T> configureChunkProvider() {
return new SimpleChunkProvider<T>(itemReader, chunkOperations);
}
/**
* Extension point for creating appropriate {@link ChunkProcessor}. Return
* value must subclass {@link SimpleChunkProcessor} due to listener
* registration.
*/
protected SimpleChunkProcessor<T, S> configureChunkProcessor() {
return new SimpleChunkProcessor<T, S>(itemProcessor, itemWriter);
}
/**
* @return a {@link CompletionPolicy} consistent with the commit interval
* and injected policy (if present).
*/
private CompletionPolicy getChunkCompletionPolicy() {
Assert.state(!(chunkCompletionPolicy != null && commitInterval != 0),
"You must specify either a chunkCompletionPolicy or a commitInterval but not both.");
Assert.state(commitInterval >= 0, "The commitInterval must be positive or zero (for default value).");
if (chunkCompletionPolicy != null) {
return chunkCompletionPolicy;
}
if (commitInterval == 0) {
logger.info("Setting commit interval to default value (" + DEFAULT_COMMIT_INTERVAL + ")");
commitInterval = DEFAULT_COMMIT_INTERVAL;
}
return new SimpleCompletionPolicy(commitInterval);
}
private void registerStreams(TaskletStep step, ItemReader<? extends T> itemReader,
ItemProcessor<? super T, ? extends S> itemProcessor, ItemWriter<? super S> itemWriter) {
for (Object itemHandler : new Object[] { itemReader, itemWriter, itemProcessor }) {
if (itemHandler instanceof ItemStream) {
registerStreams(step, new ItemStream[] { (ItemStream) itemHandler });
}
}
}
/**
* Register listeners with step and chunk.
*/
private void registerStepListeners(TaskletStep step, RepeatOperations chunkOperations) {
for (Object itemHandler : new Object[] { getItemReader(), itemWriter, itemProcessor }) {
if (StepListenerFactoryBean.isListener(itemHandler)) {
StepListener listener = StepListenerFactoryBean.getListener(itemHandler);
if (listener instanceof StepExecutionListener) {
step.registerStepExecutionListener((StepExecutionListener) listener);
}
if (listener instanceof ChunkListener) {
registerChunkListeners(step, listener);
}
}
}
step.setStepExecutionListeners(BatchListenerFactoryHelper.getListeners(listeners, StepExecutionListener.class)
.toArray(new StepExecutionListener[] {}));
List<ChunkListener> chunkListeners = BatchListenerFactoryHelper.getListeners(listeners, ChunkListener.class);
for(ChunkListener chunkListener: chunkListeners){
registerChunkListeners(step,chunkListener);
}
}
protected void registerChunkListeners(TaskletStep step, StepListener listener) {
step.registerChunkListener((ChunkListener) listener);
}
/**
* Register explicitly set ({@link #setListeners(StepListener[])}) item
* listeners and auto-register reader, processor and writer if applicable
*/
private void registerItemListeners(SimpleChunkProvider<T> chunkProvider, SimpleChunkProcessor<T, S> chunkProcessor) {
StepListener[] listeners = getListeners();
// explicitly set item listeners
chunkProvider.setListeners(BatchListenerFactoryHelper.<ItemReadListener<T>> getListeners(listeners,
ItemReadListener.class));
chunkProvider.setListeners(BatchListenerFactoryHelper.<SkipListener<T, S>> getListeners(listeners,
SkipListener.class));
chunkProcessor.setListeners(BatchListenerFactoryHelper.<ItemProcessListener<T, S>> getListeners(listeners,
ItemProcessListener.class));
chunkProcessor.setListeners(BatchListenerFactoryHelper.<ItemWriteListener<S>> getListeners(listeners,
ItemWriteListener.class));
chunkProcessor.setListeners(BatchListenerFactoryHelper.<SkipListener<T, S>> getListeners(listeners,
SkipListener.class));
List<StepListener> listofListeners = Arrays.asList(listeners);
// auto-register reader, processor and writer
for (Object itemHandler : new Object[] { getItemReader(), getItemWriter(), getItemProcessor() }) {
if (listofListeners.contains(itemHandler)) {
continue;
}
if (StepListenerFactoryBean.isListener(itemHandler)) {
StepListener listener = StepListenerFactoryBean.getListener(itemHandler);
if (listener instanceof SkipListener<?,?>) {
chunkProvider.registerListener(listener);
chunkProcessor.registerListener(listener);
// already registered with both so avoid double-registering
continue;
}
if (listener instanceof ItemReadListener<?>) {
chunkProvider.registerListener(listener);
}
if (listener instanceof ItemProcessListener<?,?> || listener instanceof ItemWriteListener<?>) {
chunkProcessor.registerListener(listener);
}
}
}
}
}

2
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java
View File

@@ -32,7 +32,7 @@ import org.springframework.batch.item.support.CompositeItemStream;
* @author Dave Syer
* @since 2.0
*/
class ChunkMonitor implements ItemStream {
public class ChunkMonitor implements ItemStream {
private Log logger = LogFactory.getLog(getClass());

56
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/OffsetItemReader.java
View File

@@ -1,56 +0,0 @@
package org.springframework.batch.core.step.item;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
/**
* Convenience wrapper for an ItemReader that keeps track of how many items
* were successfully processed.
*/
class OffsetItemReader<T> implements ItemReader<T>, ItemStream {
private static final String OFFSET_KEY = FaultTolerantStepFactoryBean.class.getName()+".OFFSET_KEY";
private final ItemReader<? extends T> itemReader;
private int offset;
/**
* @param itemReader
*/
public OffsetItemReader(ItemReader<? extends T> itemReader) {
this.itemReader = itemReader;
}
public T read() throws Exception, UnexpectedInputException, ParseException {
for (int i=0; i<offset; i++) {
// Discard items that are already processed
itemReader.read();
}
offset = 0;
return itemReader.read();
}
/**
* {@inheritDoc}
*/
public void close() throws ItemStreamException {
}
/**
* {@inheritDoc}
*/
public void open(ExecutionContext executionContext) throws ItemStreamException {
offset = executionContext.getInt(OFFSET_KEY, 0);
}
/**
* {@inheritDoc}
*/
public void update(ExecutionContext executionContext) throws ItemStreamException {
executionContext.putInt(OFFSET_KEY, offset);
}
}

2
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/ChunkElementParserTests.java
View File

@@ -316,7 +316,6 @@ public class ChunkElementParserTests {
return policy;
}
@SuppressWarnings("unchecked")
private Object getNestedPathInStep(String stepName, ApplicationContext ctx, String path) throws Exception {
Map<String, Step> beans = ctx.getBeansOfType(Step.class);
assertTrue(beans.containsKey(stepName));
@@ -365,7 +364,6 @@ public class ChunkElementParserTests {
return (Collection<ItemStream>) ReflectionTestUtils.getField(compositeStream, "streams");
}
@SuppressWarnings("unchecked")
private Collection<RetryListener> getRetryListeners(String stepName, ApplicationContext ctx) throws Exception {
Map<String, Step> beans = ctx.getBeansOfType(Step.class);
assertTrue(beans.containsKey(stepName));

11
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBeanTests.java
View File

@@ -64,6 +64,9 @@ public class StepParserStepFactoryBeanTests {
@Test
public void testOnlyTaskletSet() throws Exception {
StepParserStepFactoryBean<Object, Object> fb = new StepParserStepFactoryBean<Object, Object>();
fb.setName("step");
fb.setTransactionManager(new ResourcelessTransactionManager());
fb.setJobRepository(new JobRepositorySupport());
fb.setTasklet(new DummyTasklet());
Object step = fb.getObject();
assertTrue(step instanceof TaskletStep);
@@ -74,6 +77,9 @@ public class StepParserStepFactoryBeanTests {
@Test
public void testOnlyTaskletTaskExecutor() throws Exception {
StepParserStepFactoryBean<Object, Object> fb = new StepParserStepFactoryBean<Object, Object>();
fb.setName("step");
fb.setTransactionManager(new ResourcelessTransactionManager());
fb.setJobRepository(new JobRepositorySupport());
fb.setTasklet(new DummyTasklet());
fb.setTaskExecutor(new SimpleAsyncTaskExecutor());
Object step = fb.getObject();
@@ -85,6 +91,7 @@ public class StepParserStepFactoryBeanTests {
@Test(expected = IllegalStateException.class)
public void testSkipLimitSet() throws Exception {
StepParserStepFactoryBean<Object, Object> fb = new StepParserStepFactoryBean<Object, Object>();
fb.setName("step");
fb.setSkipLimit(5);
fb.getObject();
}
@@ -140,6 +147,7 @@ public class StepParserStepFactoryBeanTests {
fb.setItemReader(new DummyItemReader());
fb.setItemWriter(new DummyItemWriter());
fb.setStreams(new ItemStream[] { new FlatFileItemReader<Object>() });
fb.setHasChunkElement(true);
Object step = fb.getObject();
assertTrue(step instanceof TaskletStep);
@@ -147,7 +155,7 @@ public class StepParserStepFactoryBeanTests {
assertTrue(tasklet instanceof ChunkOrientedTasklet<?>);
}
@Test(expected = IllegalStateException.class)
@Test(expected = IllegalArgumentException.class)
public void testFaultTolerantStepAll() throws Exception {
StepParserStepFactoryBean<Object, Object> fb = new StepParserStepFactoryBean<Object, Object>();
fb.setBeanName("step1");
@@ -172,6 +180,7 @@ public class StepParserStepFactoryBeanTests {
fb.setRetryListeners(new RetryListenerSupport());
fb.setSkippableExceptionClasses(new HashMap<Class<? extends Throwable>, Boolean>());
fb.setRetryableExceptionClasses(new HashMap<Class<? extends Throwable>, Boolean>());
fb.setHasChunkElement(true);
Object step = fb.getObject();
assertTrue(step instanceof TaskletStep);

1
spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/PartitionStepTests.java
View File

@@ -50,6 +50,7 @@ public class PartitionStepTests {
MapJobRepositoryFactoryBean factory = new MapJobRepositoryFactoryBean();
jobRepository = (JobRepository) factory.getObject();
step.setJobRepository(jobRepository);
step.setName("partitioned");
}
@Test

43
spring-batch-core/src/test/java/org/springframework/batch/core/step/NonAbstractStepTests.java
View File

@@ -68,8 +68,7 @@ public class NonAbstractStepTests {
}
/**
* Fills the events list when listener methods are called, prefixed with the
* name of the listener.
* Fills the events list when listener methods are called, prefixed with the name of the listener.
*/
private class EventTrackingListener implements StepExecutionListener {
@@ -85,7 +84,7 @@ public class NonAbstractStepTests {
public ExitStatus afterStep(StepExecution stepExecution) {
assertSame(execution, stepExecution);
events.add(getEvent("afterStep("+stepExecution.getExitStatus().getExitCode()+")"));
events.add(getEvent("afterStep(" + stepExecution.getExitStatus().getExitCode() + ")"));
stepExecution.getExecutionContext().putString("afterStep", "afterStep");
return stepExecution.getExitStatus();
}
@@ -104,7 +103,7 @@ public class NonAbstractStepTests {
private static class JobRepositoryStub extends JobRepositorySupport {
ExecutionContext saved = new ExecutionContext();
static long counter = 0;
public void updateExecutionContext(StepExecution stepExecution) {
@@ -119,7 +118,6 @@ public class NonAbstractStepTests {
counter++;
}
}
}
@@ -128,7 +126,7 @@ public class NonAbstractStepTests {
tested.setJobRepository(repository);
repository.add(execution);
}
@Test
public void testBeanName() throws Exception {
AbstractStep step = new AbstractStep() {
@@ -175,10 +173,10 @@ public class NonAbstractStepTests {
assertEquals(ExitStatus.COMPLETED, execution.getExitStatus());
assertTrue("Execution context modifications made by listener should be persisted", repository.saved
.containsKey("beforeStep"));
assertTrue("Execution context modifications made by listener should be persisted", repository.saved
.containsKey("afterStep"));
assertTrue("Execution context modifications made by listener should be persisted",
repository.saved.containsKey("beforeStep"));
assertTrue("Execution context modifications made by listener should be persisted",
repository.saved.containsKey("afterStep"));
}
@Test
@@ -210,11 +208,12 @@ public class NonAbstractStepTests {
assertEquals(ExitStatus.FAILED.getExitCode(), execution.getExitStatus().getExitCode());
String exitDescription = execution.getExitStatus().getExitDescription();
assertTrue("Wrong message: "+exitDescription, exitDescription.contains("crash"));
assertTrue("Wrong message: " + exitDescription, exitDescription.contains("crash"));
assertTrue("Execution context modifications made by listener should be persisted", repository.saved
.containsKey("afterStep"));
assertTrue("Execution context modifications made by listener should be persisted",
repository.saved.containsKey("afterStep"));
}
/**
* Exception during business processing.
*/
@@ -247,8 +246,8 @@ public class NonAbstractStepTests {
assertEquals("STOPPED", execution.getExitStatus().getExitCode());
assertTrue("Execution context modifications made by listener should be persisted", repository.saved
.containsKey("afterStep"));
assertTrue("Execution context modifications made by listener should be persisted",
repository.saved.containsKey("afterStep"));
}
@Test
@@ -271,8 +270,8 @@ public class NonAbstractStepTests {
assertEquals("FUNNY", execution.getExitStatus().getExitCode());
assertTrue("Execution context modifications made by listener should be persisted", repository.saved
.containsKey("afterStep"));
assertTrue("Execution context modifications made by listener should be persisted",
repository.saved.containsKey("afterStep"));
}
/**
@@ -310,16 +309,10 @@ public class NonAbstractStepTests {
/**
* JobRepository is a required property.
*/
@Test
@Test(expected = IllegalStateException.class)
public void testAfterPropertiesSet() throws Exception {
tested.setJobRepository(null);
try {
tested.afterPropertiesSet();
fail();
}
catch (IllegalArgumentException e) {
// expected
}
tested.afterPropertiesSet();
}
}

1
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanNonBufferingTests.java
View File

@@ -25,6 +25,7 @@ import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.step.JobRepositorySupport;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;

1
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java
View File

@@ -45,6 +45,7 @@ import org.springframework.batch.core.repository.dao.MapJobInstanceDao;
import org.springframework.batch.core.repository.dao.MapStepExecutionDao;
import org.springframework.batch.core.repository.support.SimpleJobRepository;
import org.springframework.batch.core.step.AbstractStep;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;

2
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java
View File

@@ -30,6 +30,7 @@ import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.batch.core.step.FatalStepExecutionException;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
@@ -151,7 +152,6 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
assertFalse(attr.rollbackOn(new Exception("")));
}
@SuppressWarnings("unchecked")
@Test
public void testNonDefaultRollbackRules() throws Exception {
TransactionAttributeEditor editor = new TransactionAttributeEditor();

1
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java
View File

@@ -33,6 +33,7 @@ import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.listener.SkipListenerSupport;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
import org.springframework.batch.core.step.skip.SkipLimitExceededException;
import org.springframework.batch.core.step.skip.SkipPolicy;

1
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanUnexpectedRollbackTests.java
View File

@@ -17,6 +17,7 @@ import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.beans.factory.annotation.Autowired;

1
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBeanTests.java
View File

@@ -27,6 +27,7 @@ import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.launch.EmptyItemWriter;
import org.springframework.batch.core.step.JobRepositorySupport;
import org.springframework.batch.core.step.factory.SimpleStepFactoryBean;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.batch.repeat.RepeatCallback;
import org.springframework.batch.repeat.RepeatOperations;

3
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java
View File

@@ -47,6 +47,7 @@ import org.springframework.batch.core.repository.dao.MapJobInstanceDao;
import org.springframework.batch.core.repository.dao.MapStepExecutionDao;
import org.springframework.batch.core.repository.support.SimpleJobRepository;
import org.springframework.batch.core.step.AbstractStep;
import org.springframework.batch.core.step.factory.SimpleStepFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
@@ -272,7 +273,7 @@ public class SimpleStepFactoryBeanTests {
int expectedListenerCallCount = (items.length / commitInterval) + 1;
assertEquals(expectedListenerCallCount, chunkListener.afterCount);
assertEquals(expectedListenerCallCount, chunkListener.beforeCount);
assertTrue("Llistener order not as expected: " + writeListener.trail, writeListener.trail.startsWith("1234"));
assertTrue("Listener order not as expected: " + writeListener.trail, writeListener.trail.startsWith("1234"));
}
/**

1
spring-batch-core/src/test/resources/log4j.properties
View File

@@ -5,6 +5,7 @@ log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m%n
log4j.category.org.apache.activemq=ERROR
log4j.category.org.springframework.retry=DEBUG
log4j.category.org.springframework.batch=DEBUG
log4j.category.org.springframework.batch.support=INFO
log4j.category.org.springframework.batch.support.transaction.ResourcelessTransactionManager=DEBUG

2
spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/support/test-context.xml
View File

@@ -20,7 +20,7 @@
<bean id="test-job"
class="org.springframework.batch.core.job.JobSupport">
<property name="steps">
<bean id="step1" class="org.springframework.batch.core.step.item.SimpleStepFactoryBean">
<bean id="step1" class="org.springframework.batch.core.step.factory.SimpleStepFactoryBean">
<property name="itemReader" ref="itemReader" />
<property name="itemWriter" ref="itemWriter" />
<property name="jobRepository" ref="jobRepository" />

4
spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/2jobs.xml
View File

@@ -16,7 +16,7 @@
</property>
<property name="steps">
<bean id="step1"
class="org.springframework.batch.core.step.item.SimpleStepFactoryBean">
class="org.springframework.batch.core.step.factory.SimpleStepFactoryBean">
<property name="itemReader">
<bean
class="org.springframework.batch.item.support.ListItemReader">
@@ -46,7 +46,7 @@
</property>
<property name="steps">
<bean id="step2"
class="org.springframework.batch.core.step.item.SimpleStepFactoryBean">
class="org.springframework.batch.core.step.factory.SimpleStepFactoryBean">
<property name="itemReader">
<bean
class="org.springframework.batch.item.support.ListItemReader">

2
spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/error.xml
View File

@@ -8,7 +8,7 @@
class="org.springframework.batch.core.job.SimpleJob">
<property name="steps">
<bean id="step1"
class="org.springframework.batch.core.step.item.SimpleStepFactoryBean"/>
class="org.springframework.batch.core.step.factory.SimpleStepFactoryBean"/>
</property>
</bean>

2
spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/job.xml
View File

@@ -17,7 +17,7 @@
</property>
<property name="steps">
<bean id="step1"
class="org.springframework.batch.core.step.item.SimpleStepFactoryBean">
class="org.springframework.batch.core.step.factory.SimpleStepFactoryBean">
<property name="itemReader">
<bean class="org.springframework.batch.item.support.ListItemReader">
<constructor-arg value="foo,bar,spam" />

2
spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/job2.xml
View File

@@ -16,7 +16,7 @@
</property>
<property name="steps">
<bean id="step1"
class="org.springframework.batch.core.step.item.SimpleStepFactoryBean">
class="org.springframework.batch.core.step.factory.SimpleStepFactoryBean">
<property name="itemReader">
<bean
class="org.springframework.batch.item.support.ListItemReader">

2
spring-batch-core/src/test/resources/org/springframework/batch/core/partition/launch-context.xml
View File

@@ -58,7 +58,7 @@
</bean>
<bean id="simpleStep"
class="org.springframework.batch.core.step.item.SimpleStepFactoryBean"
class="org.springframework.batch.core.step.factory.SimpleStepFactoryBean"
abstract="true">
<property name="transactionManager" ref="transactionManager" />
<property name="jobRepository" ref="jobRepository" />

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

@@ -27,7 +27,7 @@
<property name="allowStartIfComplete" value="true" />
</bean>
<bean id="simpleStep" class="org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean" abstract="true">
<bean id="simpleStep" class="org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean" abstract="true">
<property name="transactionManager" ref="transactionManager" />
<property name="jobRepository" ref="jobRepository" />
</bean>

4
spring-batch-test/src/test/resources/org/springframework/batch/sample/config/common-context.xml
View File

@@ -18,7 +18,7 @@
<property name="allowStartIfComplete" value="true" />
</bean>
<bean id="simpleStep" class="org.springframework.batch.core.step.item.SimpleStepFactoryBean"
<bean id="simpleStep" class="org.springframework.batch.core.step.factory.SimpleStepFactoryBean"
abstract="true">
<property name="transactionManager" ref="transactionManager" />
<property name="jobRepository" ref="jobRepository" />
@@ -26,7 +26,7 @@
<property name="commitInterval" value="1" />
</bean>
<bean id="skipLimitStep" class="org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean"
<bean id="skipLimitStep" class="org.springframework.batch.core.step.factory.FaultTolerantStepFactoryBean"
parent="simpleStep" abstract="true">
<property name="skipLimit" value="0" />
</bean>