diff --git a/archetypes/simple-cli/src/main/resources/META-INF/spring/module-context.xml b/archetypes/simple-cli/src/main/resources/META-INF/spring/module-context.xml index 7e2e8b24f..81050a616 100644 --- a/archetypes/simple-cli/src/main/resources/META-INF/spring/module-context.xml +++ b/archetypes/simple-cli/src/main/resources/META-INF/spring/module-context.xml @@ -5,9 +5,12 @@ http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd"> Example job to get you started. It provides a - skeleton for a typical batch application. + skeleton for + a typical batch application. + - + @@ -17,13 +20,12 @@ - - - - + + + + diff --git a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanIntegrationTests.java b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanIntegrationTests.java index 64fd0a17c..ffc9d58e3 100644 --- a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanIntegrationTests.java +++ b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanIntegrationTests.java @@ -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; diff --git a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackIntegrationTests.java b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackIntegrationTests.java index 2ba066f78..c8a290b75 100644 --- a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackIntegrationTests.java +++ b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackIntegrationTests.java @@ -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; diff --git a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/MapRepositoryFaultTolerantStepFactoryBeanRollbackTests.java b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/MapRepositoryFaultTolerantStepFactoryBeanRollbackTests.java index 033547deb..cfd47fa1a 100644 --- a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/MapRepositoryFaultTolerantStepFactoryBeanRollbackTests.java +++ b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/MapRepositoryFaultTolerantStepFactoryBeanRollbackTests.java @@ -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; diff --git a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/MapRepositoryFaultTolerantStepFactoryBeanTests.java b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/MapRepositoryFaultTolerantStepFactoryBeanTests.java index 58ca3a330..6044d1f13 100644 --- a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/MapRepositoryFaultTolerantStepFactoryBeanTests.java +++ b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/MapRepositoryFaultTolerantStepFactoryBeanTests.java @@ -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; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java index 1600f8ba9..d62486d2a 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java @@ -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 <step/> (and its inner <tasklet/>). 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 <step/> (and its inner <tasklet/>). 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 implements FactoryBean, BeanNameAware { - private static final Log logger = LogFactory.getLog(StepParserStepFactoryBean.class); - // // Step Attributes // @@ -109,6 +108,8 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { private PlatformTransactionManager transactionManager; + private Set stepExecutionListeners = new LinkedHashSet(); + // // Flow Elements // @@ -139,8 +140,6 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { // // Tasklet Elements // - private StepListener[] listeners; - private Collection> noRollbackExceptionClasses; private Integer transactionTimeout; @@ -149,10 +148,12 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { private Isolation isolation; + private Set chunkListeners = new LinkedHashSet(); + // // Chunk Attributes // - private Integer cacheCapacity; + private int cacheCapacity = 0; private CompletionPolicy chunkCompletionPolicy; @@ -162,7 +163,7 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { private Boolean processorTransactional; - private Integer retryLimit; + private int retryLimit = 0; private BackOffPolicy backOffPolicy; @@ -191,12 +192,20 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { // private RetryListener[] retryListeners; - private Map, Boolean> skippableExceptionClasses; + private Map, Boolean> skippableExceptionClasses = new HashMap, Boolean>(); - private Map, Boolean> retryableExceptionClasses; + private Map, Boolean> retryableExceptionClasses = new HashMap, Boolean>(); private ItemStream[] streams; + private Set> readListeners = new LinkedHashSet>(); + + private Set> writeListeners = new LinkedHashSet>(); + + private Set> processListeners = new LinkedHashSet>(); + + private Set> skipListeners = new LinkedHashSet>(); + // // Additional // @@ -204,6 +213,8 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { private StepExecutionAggregator stepExecutionAggregator; + private StepListener[] listeners; + /** * Create a {@link Step} from the configuration provided. * @@ -216,36 +227,23 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { validateFaultTolerantSettings(); if (isFaultTolerant()) { - FaultTolerantStepFactoryBean fb = new FaultTolerantStepFactoryBean(); - configureSimple(fb); - configureFaultTolerant(fb); - return fb.getObject(); + return createFaultTolerantStep(); } else { - SimpleStepFactoryBean fb = new SimpleStepFactoryBean(); - 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 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 newListeners = new ArrayList(); - 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 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 builder = new FaultTolerantStepBuilder(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 listener : skipListeners) { + builder.listener(listener); } - } - private void configureFaultTolerant(FaultTolerantStepFactoryBean 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 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 type : retryableExceptionClasses.keySet()) { + if (retryableExceptionClasses.get(type)) { + builder.retry(type); + } + else { + builder.noRetry(type); + } + } + } + if (noRollbackExceptionClasses != null) { - fb.setNoRollbackExceptionClasses(noRollbackExceptionClasses); + for (Class type : noRollbackExceptionClasses) { + builder.noRollback(type); + } + } + + return builder.build(); + + } + + private void registerItemListeners(SimpleStepBuilder builder) { + for (ItemReadListener listener : readListeners) { + builder.listener(listener); + } + for (ItemWriteListener listener : writeListeners) { + builder.listener(listener); + } + for (ItemProcessListener listener : processListeners) { + builder.listener(listener); } } - @SuppressWarnings("serial") - private void configureTaskletStep(TaskletStep ts) { - configureAbstractStep(ts); - if (listeners != null) { - List newListeners = new ArrayList(); - 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 implements FactoryBean, BeanNameAware { Collection> exceptions = noRollbackExceptionClasses == null ? new HashSet>() : 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 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 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 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 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 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 skipListener = (SkipListener) 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 readListener = (ItemReadListener) listener; + readListeners.add(readListener); + } + if (listener instanceof ItemWriteListener) { + @SuppressWarnings("unchecked") + ItemWriteListener writeListener = (ItemWriteListener) listener; + writeListeners.add(writeListener); + } + if (listener instanceof ItemProcessListener) { + @SuppressWarnings("unchecked") + ItemProcessListener processListener = (ItemProcessListener) 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 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 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 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.
*

- * 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}.
* - * @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 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 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 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 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 retryLimit == 1 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 retryLimit == 1 by default. * * @param retryLimit the retry limit to set, must be greater or equal to 1. */ @@ -902,11 +868,9 @@ class StepParserStepFactoryBean 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 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 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 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 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 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 */ diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowStep.java index 466a1555c..071f39577 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowStep.java @@ -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(); } /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java index c8006c1cb..79e78e189 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java @@ -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} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/AbstractTaskletStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/AbstractTaskletStepBuilder.java new file mode 100644 index 000000000..f840b0ff5 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/AbstractTaskletStepBuilder.java @@ -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> extends + StepBuilderHelper> { + + private Set listeners = new LinkedHashSet(); + + private RepeatOperations stepOperations; + + private TransactionAttribute transactionAttribute; + + private Set streams = new LinkedHashSet(); + + 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 listener(ChunkListener listener) { + listeners.add(listener); + return this; + } + + public AbstractTaskletStepBuilder stream(ItemStream stream) { + streams.add(stream); + return this; + } + + public AbstractTaskletStepBuilder taskExecutor(TaskExecutor taskExecutor) { + this.taskExecutor = taskExecutor; + return this; + } + + public AbstractTaskletStepBuilder throttleLimit(int throttleLimit) { + this.throttleLimit = throttleLimit; + return this; + } + + public AbstractTaskletStepBuilder exceptionHandler(ExceptionHandler exceptionHandler) { + this.exceptionHandler = exceptionHandler; + return this; + } + + public AbstractTaskletStepBuilder stepOperations(RepeatOperations repeatTemplate) { + this.stepOperations = repeatTemplate; + return this; + } + + public AbstractTaskletStepBuilder 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; + } + +} \ No newline at end of file diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java old mode 100755 new mode 100644 similarity index 53% rename from spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java rename to spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java index 1b276d603..4f46342eb --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java @@ -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 extends SimpleStepFactoryBean { +public class FaultTolerantStepBuilder extends SimpleStepBuilder { - private Map, Boolean> skippableExceptionClasses = new HashMap, Boolean>(); + private ChunkMonitor chunkMonitor = new ChunkMonitor(); - private Collection> nonSkippableExceptionClasses = new HashSet>(); - - private Collection> noRollbackExceptionClasses = new HashSet>(); - - private Map, Boolean> retryableExceptionClasses = new HashMap, Boolean>(); - - private Collection> nonRetryableExceptionClasses = new HashSet>(); - - 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 retryListeners = new LinkedHashSet(); private RetryPolicy retryPolicy; @@ -113,182 +92,235 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean> noRollbackExceptionClasses = new LinkedHashSet>(); + + private Map, Boolean> skippableExceptionClasses = new HashMap, Boolean>(); + + private Collection> nonSkippableExceptionClasses = new HashSet>(); + + private Map, Boolean> retryableExceptionClasses = new HashMap, Boolean>(); + + private Collection> nonRetryableExceptionClasses = new HashSet>(); + + private Set> skipListeners = new LinkedHashSet>(); + + 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 chunkProvider = createChunkProvider(); + FaultTolerantChunkProcessor chunkProcessor = createChunkProcessor(); + ChunkOrientedTasklet tasklet = new ChunkOrientedTasklet(chunkProvider, chunkProcessor); + tasklet.setBuffering(!isReaderTransactionalQueue()); + return tasklet; + } + + public FaultTolerantStepBuilder listener(SkipListener listener) { + skipListeners.add(listener); + return this; + } + + @Override + public FaultTolerantStepBuilder listener(ChunkListener listener) { + super.listener(new TerminateOnExceptionChunkListenerDelegate(listener)); + return this; + } + + @Override + public AbstractTaskletStepBuilder> transactionAttribute( + TransactionAttribute transactionAttribute) { + return super.transactionAttribute(getTransactionAttribute(transactionAttribute)); + } + + public FaultTolerantStepBuilder listener(RetryListener listener) { + retryListeners.add(listener); + return this; + } + + public FaultTolerantStepBuilder 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 retryLimit == 1 by default. - * - * @param retryLimit - * the retry limit to set, must be greater or equal to 1. - */ - public void setRetryLimit(int retryLimit) { + + public FaultTolerantStepBuilder 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.
- * - * 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}.
- * - * 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 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, 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 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 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 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 skipPolicy(SkipPolicy skipPolicy) { this.skipPolicy = skipPolicy; + return this; + } + + public FaultTolerantStepBuilder noRollback(Class type) { + noRollbackExceptionClasses.add(type); + return this; + } + + public FaultTolerantStepBuilder noRetry(Class type) { + retryableExceptionClasses.put(type, false); + return this; + } + + public FaultTolerantStepBuilder retry(Class type) { + retryableExceptionClasses.put(type, true); + return this; + } + + public FaultTolerantStepBuilder noSkip(Class type) { + skippableExceptionClasses.put(type, false); + return this; + } + + public FaultTolerantStepBuilder skip(Class type) { + skippableExceptionClasses.put(type, true); + return this; + } + + public FaultTolerantStepBuilder processorNonTransactional() { + this.processorTransactional = false; + return this; + } + + @Override + public AbstractTaskletStepBuilder> 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 createChunkProvider() { + + SkipPolicy readSkipPolicy = createSkipPolicy(); + readSkipPolicy = getFatalExceptionAwareProxy(readSkipPolicy); + FaultTolerantChunkProvider chunkProvider = new FaultTolerantChunkProvider(getReader(), + createChunkOperations()); + chunkProvider.setMaxSkipsOnRead(Math.max(getChunkSize(), FaultTolerantChunkProvider.DEFAULT_MAX_SKIPS_ON_READ)); + chunkProvider.setSkipPolicy(readSkipPolicy); + chunkProvider.setRollbackClassifier(getRollbackClassifier()); + ArrayList listeners = new ArrayList(getItemListeners()); + listeners.addAll(skipListeners); + chunkProvider.setListeners(listeners); + + return chunkProvider; + + } + + private FaultTolerantChunkProcessor createChunkProcessor() { + + BatchRetryTemplate batchRetryTemplate = createRetryOperations(); + + FaultTolerantChunkProcessor chunkProcessor = new FaultTolerantChunkProcessor(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 listeners = new ArrayList(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. - *

- * 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 skipListener = (SkipListener) listener; + skipListeners.add(skipListener); + } + } + + } + } + + /** + * Convenience method to get an exception classifier based on the provided transaction attributes. * - * @param exceptionClasses - * defaults to Exception + * @return an exception classifier: maps to true if an exception should cause rollback */ - public void setSkippableExceptionClasses(Map, 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. - *

- * Defaults is empty. - * - * @param noRollbackExceptionClasses - * the exception classes to set - */ - public void setNoRollbackExceptionClasses(Collection> 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 getRollbackClassifier() { + private Classifier getRollbackClassifier() { Classifier classifier = new BinaryExceptionClassifier(noRollbackExceptionClasses, false); @@ -317,16 +349,8 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean classifier = getRollbackClassifier(); return new DefaultTransactionAttribute(attribute) { @Override @@ -338,118 +362,33 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean 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 configureChunkProvider() { - - SkipPolicy readSkipPolicy = createSkipPolicy(); - readSkipPolicy = getFatalExceptionAwareProxy(readSkipPolicy); - FaultTolerantChunkProvider chunkProvider = new FaultTolerantChunkProvider(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, Boolean> map = new HashMap, Boolean>(skippableExceptionClasses); + Map, Boolean> map = new HashMap, 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 configureChunkProcessor() { - - BatchRetryTemplate batchRetryTemplate = configureRetry(); - - FaultTolerantChunkProcessor chunkProcessor = new FaultTolerantChunkProcessor(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, Boolean> map = new HashMap, Boolean>(retryableExceptionClasses); + Map, Boolean> map = new HashMap, Boolean>( + retryableExceptionClasses); map.put(ForceRollbackForWriteSkipException.class, true); simpleRetryPolicy = new SimpleRetryPolicy(retryLimit, map); @@ -457,7 +396,8 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean 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 extends SimpleStepFactoryBean 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 extends SimpleStepFactoryBean classifier = new SubclassClassifier(retryPolicy); + SubclassClassifier classifier = new SubclassClassifier( + retryPolicy); classifier.setTypeMap(map); ExceptionClassifierRetryPolicy retryPolicyWrapper = new ExceptionClassifierRetryPolicy(); @@ -515,11 +453,9 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean extends SimpleStepFactoryBean>) 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 extends SimpleStepFactoryBean extends SimpleStepFactoryBean { + + 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; + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/JobStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/JobStepBuilder.java new file mode 100644 index 000000000..66e8411cf --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/JobStepBuilder.java @@ -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 { + + 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; + + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/PartitionStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/PartitionStepBuilder.java new file mode 100644 index 000000000..47b9ce7d1 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/PartitionStepBuilder.java @@ -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 { + + 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; + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java new file mode 100644 index 000000000..b92913685 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java @@ -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 extends AbstractTaskletStepBuilder> { + + private static final int DEFAULT_COMMIT_INTERVAL = 1; + + private ItemReader reader; + + private ItemWriter writer; + + private ItemProcessor processor; + + private int chunkSize = 0; + + private RepeatOperations chunkOperations; + + private CompletionPolicy completionPolicy; + + private Set itemListeners = new LinkedHashSet(); + + 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 chunkProvider = new SimpleChunkProvider(reader, repeatOperations); + SimpleChunkProcessor chunkProcessor = new SimpleChunkProcessor(processor, writer); + chunkProvider.setListeners(new ArrayList(itemListeners)); + chunkProcessor.setListeners(new ArrayList(itemListeners)); + ChunkOrientedTasklet tasklet = new ChunkOrientedTasklet(chunkProvider, chunkProcessor); + tasklet.setBuffering(!readerTransactionalQueue); + return tasklet; + } + + public SimpleStepBuilder readerIsTransactionalQueue() { + this.readerTransactionalQueue = true; + return this; + } + + public SimpleStepBuilder listener(ItemReadListener listener) { + itemListeners.add(listener); + return this; + } + + public SimpleStepBuilder listener(ItemWriteListener listener) { + itemListeners.add(listener); + return this; + } + + public SimpleStepBuilder listener(ItemProcessListener listener) { + itemListeners.add(listener); + return this; + } + + public SimpleStepBuilder chunkOperations(RepeatOperations repeatTemplate) { + this.chunkOperations = repeatTemplate; + return this; + } + + public SimpleStepBuilder 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 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 chunk(CompletionPolicy completionPolicy) { + this.completionPolicy = completionPolicy; + return this; + } + + public SimpleStepBuilder reader(ItemReader reader) { + this.reader = reader; + return this; + } + + public SimpleStepBuilder writer(ItemWriter writer) { + this.writer = writer; + return this; + } + + public SimpleStepBuilder processor(ItemProcessor 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 getReader() { + return reader; + } + + protected ItemWriter getWriter() { + return writer; + } + + protected ItemProcessor getProcessor() { + return processor; + } + + protected int getChunkSize() { + return chunkSize; + } + + protected boolean isReaderTransactionalQueue() { + return readerTransactionalQueue; + } + + protected Set 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 itemReader, + ItemProcessor itemProcessor, ItemWriter 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); + } + } + } + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilder.java new file mode 100644 index 000000000..ed31cb85d --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilder.java @@ -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 { + + public StepBuilder(String name) { + super(name); + } + + public TaskletStepBuilder tasklet(Tasklet tasklet) { + return new TaskletStepBuilder(this).tasklet(tasklet); + } + + public SimpleStepBuilder chunk(int chunkSize) { + return new SimpleStepBuilder(this).chunk(chunkSize); + } + + public SimpleStepBuilder chunk(CompletionPolicy completionPolicy) { + return new SimpleStepBuilder(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); + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderException.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderException.java new file mode 100644 index 000000000..0538d3e54 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderException.java @@ -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); + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderHelper.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderHelper.java new file mode 100644 index 000000000..bb95db50f --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderHelper.java @@ -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> { + + 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 repository(JobRepository jobRepository) { + properties.jobRepository = jobRepository; + return this; + } + + public StepBuilderHelper transactionManager(PlatformTransactionManager transactionManager) { + properties.transactionManager = transactionManager; + return this; + } + + public StepBuilderHelper startLimit(int startLimit) { + properties.startLimit = startLimit; + return this; + } + + public StepBuilderHelper listener(StepExecutionListener listener) { + properties.addStepExecutionListener(listener); + return this; + } + + public StepBuilderHelper 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 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 stepExecutionListeners = new ArrayList(); + + 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 getStepExecutionListeners() { + return stepExecutionListeners; + } + + public void addStepExecutionListeners(List 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; + + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/TaskletStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/TaskletStepBuilder.java new file mode 100644 index 000000000..9289dd4cc --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/TaskletStepBuilder.java @@ -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 { + + 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; + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/factory/BatchListenerFactoryHelper.java similarity index 89% rename from spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java rename to spring-batch-core/src/main/java/org/springframework/batch/core/step/factory/BatchListenerFactoryHelper.java index e336c78fd..72a5ebe1b 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/factory/BatchListenerFactoryHelper.java @@ -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 List getListeners(StepListener[] listeners, Class cls) { List list = new ArrayList(); - 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; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/factory/FaultTolerantStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/factory/FaultTolerantStepFactoryBean.java new file mode 100755 index 000000000..8cee29a06 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/factory/FaultTolerantStepFactoryBean.java @@ -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 extends SimpleStepFactoryBean { + + private Map, Boolean> skippableExceptionClasses = new HashMap, Boolean>(); + + private Collection> noRollbackExceptionClasses = new HashSet>(); + + private Map, Boolean> retryableExceptionClasses = new HashMap, 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 retryLimit == 1 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.
+ * + * 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}.
+ * + * 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, 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. + *

+ * Defaults to all no exception. + * + * @param exceptionClasses defaults to Exception + */ + public void setSkippableExceptionClasses(Map, 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. + *

+ * Defaults is empty. + * + * @param noRollbackExceptionClasses the exception classes to set + */ + public void setNoRollbackExceptionClasses(Collection> noRollbackExceptionClasses) { + this.noRollbackExceptionClasses = noRollbackExceptionClasses; + } + + /** + * @param processorTransactional + */ + public void setProcessorTransactional(boolean processorTransactional) { + this.processorTransactional = processorTransactional; + } + + protected SimpleStepBuilder createBuilder(String name) { + return new FaultTolerantStepBuilder(new StepBuilder(name)); + } + + @Override + protected void applyConfiguration(SimpleStepBuilder builder) { + + FaultTolerantStepBuilder faultTolerantBuilder = (FaultTolerantStepBuilder) builder; + + if (retryContextCache == null && cacheCapacity > 0) { + retryContextCache = new MapRetryContextCache(cacheCapacity); + } + faultTolerantBuilder.retryContextCache(retryContextCache); + for (SkipListener listener : BatchListenerFactoryHelper.> 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 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 type : retryableExceptionClasses.keySet()) { + if (retryableExceptionClasses.get(type)) { + faultTolerantBuilder.retry(type); + } + else { + faultTolerantBuilder.noRetry(type); + } + } + + for (Class type : noRollbackExceptionClasses) { + faultTolerantBuilder.noRollback(type); + } + super.applyConfiguration(builder); + + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/factory/SimpleStepFactoryBean.java similarity index 57% rename from spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java rename to spring-batch-core/src/main/java/org/springframework/batch/core/step/factory/SimpleStepFactoryBean.java index 5d589f252..414cc730e 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/factory/SimpleStepFactoryBean.java @@ -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 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 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 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 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 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 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 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 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 implements FactoryBean, BeanNameAware { * @see FactoryBean#getObject() */ public final Object getObject() throws Exception { - TaskletStep step = new TaskletStep(getName()); - applyConfiguration(step); - step.afterPropertiesSet(); + SimpleStepBuilder builder = createBuilder(getName()); + applyConfiguration(builder); + TaskletStep step = builder.build(); return step; } + protected SimpleStepBuilder createBuilder(String name) { + return new SimpleStepBuilder(new StepBuilder(name)); + } + public Class 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 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 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 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 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 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 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. getListeners( + listeners, StepExecutionListener.class)) { + builder.listener(listener); + } + for (ChunkListener listener : BatchListenerFactoryHelper. getListeners(listeners, + ChunkListener.class)) { + builder.listener(listener); + } + for (ItemReadListener listener : BatchListenerFactoryHelper.> getListeners(listeners, + ItemReadListener.class)) { + builder.listener(listener); + } + for (ItemWriteListener listener : BatchListenerFactoryHelper.> getListeners(listeners, + ItemWriteListener.class)) { + builder.listener(listener); + } + for (ItemProcessListener listener : BatchListenerFactoryHelper.> 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 chunkProvider = configureChunkProvider(); - - SimpleChunkProcessor chunkProcessor = configureChunkProcessor(); - - registerItemListeners(chunkProvider, chunkProcessor); - registerStepListeners(step, chunkOperations); - registerStreams(step, itemReader, itemProcessor, itemWriter); - - ChunkOrientedTasklet tasklet = new ChunkOrientedTasklet(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 configureChunkProvider() { - return new SimpleChunkProvider(itemReader, chunkOperations); - } - - /** - * Extension point for creating appropriate {@link ChunkProcessor}. Return - * value must subclass {@link SimpleChunkProcessor} due to listener - * registration. - */ - protected SimpleChunkProcessor configureChunkProcessor() { - return new SimpleChunkProcessor(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 itemReader, - ItemProcessor itemProcessor, ItemWriter 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 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 chunkProvider, SimpleChunkProcessor chunkProcessor) { - - StepListener[] listeners = getListeners(); - - // explicitly set item listeners - chunkProvider.setListeners(BatchListenerFactoryHelper.> getListeners(listeners, - ItemReadListener.class)); - chunkProvider.setListeners(BatchListenerFactoryHelper.> getListeners(listeners, - SkipListener.class)); - - chunkProcessor.setListeners(BatchListenerFactoryHelper.> getListeners(listeners, - ItemProcessListener.class)); - chunkProcessor.setListeners(BatchListenerFactoryHelper.> getListeners(listeners, - ItemWriteListener.class)); - chunkProcessor.setListeners(BatchListenerFactoryHelper.> getListeners(listeners, - SkipListener.class)); - - List 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); - } - } - } - } - } \ No newline at end of file diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java index 97b669f38..684ab57b1 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java @@ -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()); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/OffsetItemReader.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/OffsetItemReader.java deleted file mode 100644 index 70beae803..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/OffsetItemReader.java +++ /dev/null @@ -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 implements ItemReader, ItemStream { - - private static final String OFFSET_KEY = FaultTolerantStepFactoryBean.class.getName()+".OFFSET_KEY"; - private final ItemReader itemReader; - private int offset; - - /** - * @param itemReader - */ - public OffsetItemReader(ItemReader itemReader) { - this.itemReader = itemReader; - } - - public T read() throws Exception, UnexpectedInputException, ParseException { - for (int i=0; i beans = ctx.getBeansOfType(Step.class); assertTrue(beans.containsKey(stepName)); @@ -365,7 +364,6 @@ public class ChunkElementParserTests { return (Collection) ReflectionTestUtils.getField(compositeStream, "streams"); } - @SuppressWarnings("unchecked") private Collection getRetryListeners(String stepName, ApplicationContext ctx) throws Exception { Map beans = ctx.getBeansOfType(Step.class); assertTrue(beans.containsKey(stepName)); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBeanTests.java index 744068aa4..218863208 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBeanTests.java @@ -64,6 +64,9 @@ public class StepParserStepFactoryBeanTests { @Test public void testOnlyTaskletSet() throws Exception { StepParserStepFactoryBean fb = new StepParserStepFactoryBean(); + 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 fb = new StepParserStepFactoryBean(); + 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 fb = new StepParserStepFactoryBean(); + 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() }); + 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 fb = new StepParserStepFactoryBean(); fb.setBeanName("step1"); @@ -172,6 +180,7 @@ public class StepParserStepFactoryBeanTests { fb.setRetryListeners(new RetryListenerSupport()); fb.setSkippableExceptionClasses(new HashMap, Boolean>()); fb.setRetryableExceptionClasses(new HashMap, Boolean>()); + fb.setHasChunkElement(true); Object step = fb.getObject(); assertTrue(step instanceof TaskletStep); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/PartitionStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/PartitionStepTests.java index 3f1e2fbcc..fe8250c9e 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/PartitionStepTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/PartitionStepTests.java @@ -50,6 +50,7 @@ public class PartitionStepTests { MapJobRepositoryFactoryBean factory = new MapJobRepositoryFactoryBean(); jobRepository = (JobRepository) factory.getObject(); step.setJobRepository(jobRepository); + step.setName("partitioned"); } @Test diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/NonAbstractStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/NonAbstractStepTests.java index 8071312e4..9929618cc 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/NonAbstractStepTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/NonAbstractStepTests.java @@ -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(); } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanNonBufferingTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanNonBufferingTests.java index dc75b1570..6a6d327ca 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanNonBufferingTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanNonBufferingTests.java @@ -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; diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java index ec204823f..baa9498ec 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java @@ -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; diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java index c5dc4d120..2799fcea0 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java @@ -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(); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java index 552028ca3..ee7ce37bd 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java @@ -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; diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanUnexpectedRollbackTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanUnexpectedRollbackTests.java index 933c4484b..1bb17378d 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanUnexpectedRollbackTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanUnexpectedRollbackTests.java @@ -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; diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBeanTests.java index 2f29789ab..b65728232 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBeanTests.java @@ -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; diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java index bb44e6ac9..428b02853 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java @@ -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")); } /** diff --git a/spring-batch-core/src/test/resources/log4j.properties b/spring-batch-core/src/test/resources/log4j.properties index 3cfece9d8..0d7d45982 100644 --- a/spring-batch-core/src/test/resources/log4j.properties +++ b/spring-batch-core/src/test/resources/log4j.properties @@ -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 diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/support/test-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/support/test-context.xml index 4d245b2ff..617b89cfe 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/support/test-context.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/support/test-context.xml @@ -20,7 +20,7 @@ - + diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/2jobs.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/2jobs.xml index 7cf6c842e..298091fae 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/2jobs.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/2jobs.xml @@ -16,7 +16,7 @@ + class="org.springframework.batch.core.step.factory.SimpleStepFactoryBean"> @@ -46,7 +46,7 @@ + class="org.springframework.batch.core.step.factory.SimpleStepFactoryBean"> diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/error.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/error.xml index 059347104..260dcfa02 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/error.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/error.xml @@ -8,7 +8,7 @@ class="org.springframework.batch.core.job.SimpleJob"> + class="org.springframework.batch.core.step.factory.SimpleStepFactoryBean"/> diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/job.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/job.xml index a309e7a3c..a60094b69 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/job.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/job.xml @@ -17,7 +17,7 @@ + class="org.springframework.batch.core.step.factory.SimpleStepFactoryBean"> diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/job2.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/job2.xml index 1586a1a03..cbcb6729e 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/job2.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/launch/support/job2.xml @@ -16,7 +16,7 @@ + class="org.springframework.batch.core.step.factory.SimpleStepFactoryBean"> diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/partition/launch-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/partition/launch-context.xml index 585d51331..a02739b93 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/partition/launch-context.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/partition/launch-context.xml @@ -58,7 +58,7 @@ diff --git a/spring-batch-samples/src/main/resources/jobs/infiniteLoopJob.xml b/spring-batch-samples/src/main/resources/jobs/infiniteLoopJob.xml index ae80b57a6..560935156 100644 --- a/spring-batch-samples/src/main/resources/jobs/infiniteLoopJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/infiniteLoopJob.xml @@ -27,7 +27,7 @@ - + diff --git a/spring-batch-test/src/test/resources/org/springframework/batch/sample/config/common-context.xml b/spring-batch-test/src/test/resources/org/springframework/batch/sample/config/common-context.xml index ba377b0de..d725c26e6 100755 --- a/spring-batch-test/src/test/resources/org/springframework/batch/sample/config/common-context.xml +++ b/spring-batch-test/src/test/resources/org/springframework/batch/sample/config/common-context.xml @@ -18,7 +18,7 @@ - @@ -26,7 +26,7 @@ -