diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AbstractStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AbstractStep.java index d89d85034..eeefe6f59 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AbstractStep.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AbstractStep.java @@ -20,11 +20,14 @@ import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.core.domain.StepInterruptedException; import org.springframework.batch.core.domain.StepSupport; import org.springframework.batch.core.repository.JobRepository; -import org.springframework.batch.core.tasklet.Tasklet; import org.springframework.batch.io.exception.BatchCriticalException; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemRecoverer; +import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.stream.SimpleStreamManager; import org.springframework.batch.item.stream.StreamManager; import org.springframework.batch.repeat.exception.handler.ExceptionHandler; +import org.springframework.beans.factory.InitializingBean; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.util.Assert; @@ -32,9 +35,9 @@ import org.springframework.util.Assert; * A {@link Step} implementation that provides common behaviour to subclasses. * * @author Dave Syer - * + * @author Ben Hale */ -public abstract class AbstractStep extends StepSupport { +public abstract class AbstractStep extends StepSupport implements InitializingBean { private int skipLimit = 0; @@ -44,10 +47,14 @@ public abstract class AbstractStep extends StepSupport { private PlatformTransactionManager transactionManager; - private Tasklet tasklet; - private StreamManager streamManager; + private ItemReader itemReader; + + private ItemWriter itemWriter; + + private ItemRecoverer itemRecoverer; + /** * Default constructor. */ @@ -57,6 +64,7 @@ public abstract class AbstractStep extends StepSupport { /** * Convenient constructor for setting only the name property. + * * @param name */ public AbstractStep(String name) { @@ -90,6 +98,7 @@ public abstract class AbstractStep extends StepSupport { /** * Public setter for the {@link PlatformTransactionManager}. + * * @param transactionManager the transaction manager to set */ public void setTransactionManager(PlatformTransactionManager transactionManager) { @@ -97,14 +106,35 @@ public abstract class AbstractStep extends StepSupport { } /** - * Public setter for the {@link StreamManager}. Set either this or the - * transaction manager, but not both. + * Public setter for the {@link StreamManager}. Set either this or the transaction manager, but not both. + * * @param streamManager the {@link StreamManager} to set. */ public void setStreamManager(StreamManager streamManager) { this.streamManager = streamManager; } + /** + * @param itemReader the itemReader to set + */ + public void setItemReader(ItemReader itemReader) { + this.itemReader = itemReader; + } + + /** + * @param itemWriter the itemWriter to set + */ + public void setItemWriter(ItemWriter itemWriter) { + this.itemWriter = itemWriter; + } + + /** + * @param itemRecoverer the itemRecoverer to set + */ + public void setItemRecoverer(ItemRecoverer itemRecoverer) { + this.itemRecoverer = itemRecoverer; + } + /** * Assert that all mandatory properties are set (the {@link JobRepository}). * @@ -117,33 +147,24 @@ public abstract class AbstractStep extends StepSupport { protected void assertMandatoryProperties() { Assert.notNull(jobRepository, "JobRepository is mandatory"); Assert.state(transactionManager != null || streamManager != null, - "Either StreamManager or TransactionManager must be set"); + "Either StreamManager or TransactionManager must be set"); Assert.state(transactionManager == null || streamManager == null, - "Only one of StreamManager or TransactionManager must be set"); + "Only one of StreamManager or TransactionManager must be set"); + Assert.notNull(itemReader, "ItemReader must be provided"); + Assert.notNull(itemWriter, "ItemWriter must be provided"); + } - /* - * (non-Javadoc) - * @see org.springframework.batch.core.domain.StepSupport#process(org.springframework.batch.core.domain.StepExecution) - */ public void execute(StepExecution stepExecution) throws StepInterruptedException, BatchCriticalException { SimpleStepExecutor executor = createStepExecutor(); executor.execute(stepExecution); } - /** - * Public setter for the tasklet. - * - * @param tasklet the tasklet to set - */ - public void setTasklet(Tasklet tasklet) { - this.tasklet = tasklet; - } - /** * @return a {@link SimpleStepExecutor} that can be used to launch the job. + * @throws BatchCriticalException */ - protected SimpleStepExecutor createStepExecutor() { + protected SimpleStepExecutor createStepExecutor() throws BatchCriticalException { assertMandatoryProperties(); // Do not set the streamManager field if it is null, otherwise // the mandatory properties check will fail. @@ -152,11 +173,17 @@ public abstract class AbstractStep extends StepSupport { manager = new SimpleStreamManager(transactionManager); } SimpleStepExecutor executor = new SimpleStepExecutor(this); + executor.setItemReader(itemReader); + executor.setItemWriter(itemWriter); + executor.setItemRecoverer(itemRecoverer); executor.setRepository(jobRepository); executor.setStreamManager(manager); + try { + executor.afterPropertiesSet(); + } catch (Exception e) { + throw new BatchCriticalException(e); + } executor.applyConfiguration(this); - executor.setTasklet(tasklet); return executor; } - } \ No newline at end of file diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/RepeatOperationsStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/RepeatOperationsStep.java index bdc046cf5..8c1b898b5 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/RepeatOperationsStep.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/RepeatOperationsStep.java @@ -19,6 +19,7 @@ package org.springframework.batch.execution.step.simple; import org.springframework.batch.core.domain.Step; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.core.domain.StepInterruptedException; +import org.springframework.batch.core.tasklet.Tasklet; import org.springframework.batch.io.exception.BatchCriticalException; import org.springframework.batch.repeat.RepeatOperations; @@ -28,15 +29,15 @@ import org.springframework.batch.repeat.RepeatOperations; * * @author Lucas Ward * @author Dave Syer - * + * @author Ben Hale */ public class RepeatOperationsStep extends AbstractStep implements RepeatOperationsHolder { - // default chunkOperations is null - private RepeatOperations chunkOperations; + private volatile RepeatOperations chunkOperations; - // default stepOperations is null - private RepeatOperations stepOperations; + private volatile RepeatOperations stepOperations; + + private volatile Tasklet tasklet; /** * Public accessor for the chunkOperations property. @@ -55,6 +56,11 @@ public class RepeatOperationsStep extends AbstractStep implements RepeatOperatio public void setChunkOperations(RepeatOperations chunkOperations) { this.chunkOperations = chunkOperations; } + + + public void setTasklet(Tasklet tasklet) { + this.tasklet = tasklet; + } /** * Public accessor for the stepOperations property. @@ -74,9 +80,6 @@ public class RepeatOperationsStep extends AbstractStep implements RepeatOperatio this.stepOperations = stepOperations; } - /* (non-Javadoc) - * @see org.springframework.batch.execution.step.simple.AbstractStep#process(org.springframework.batch.core.domain.StepExecution) - */ public void execute(StepExecution stepExecution) throws StepInterruptedException, BatchCriticalException { assertMandatoryProperties(); SimpleStepExecutor executor = (SimpleStepExecutor) super.createStepExecutor(); diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStep.java index 376f126c6..be9e4dd1e 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStep.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStep.java @@ -17,20 +17,20 @@ package org.springframework.batch.execution.step.simple; import org.springframework.batch.core.domain.Step; -import org.springframework.batch.core.tasklet.Tasklet; /** - * Simple {@link Step} good enough for most purposes and easy to - * configure simple properties, principally the commit interval. + * Simple {@link Step} good enough for most purposes and easy to configure simple properties, principally the commit + * interval. * * @author Lucas Ward * @author Dave Syer - * + * @author Ben Hale */ public class SimpleStep extends AbstractStep { - + // default commit interval is one private int commitInterval = 1; + public SimpleStep() { super(); } @@ -39,11 +39,6 @@ public class SimpleStep extends AbstractStep { super(name); } - public SimpleStep(Tasklet module) { - this(); - setTasklet(module); - } - public void setCommitInterval(int commitInterval) { this.commitInterval = commitInterval; } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java index 4a8ba3323..f309aa8ee 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java @@ -13,7 +13,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.springframework.batch.execution.step.simple; import java.util.Date; @@ -36,7 +35,11 @@ import org.springframework.batch.execution.scope.StepSynchronizationManager; import org.springframework.batch.io.Skippable; import org.springframework.batch.io.exception.BatchCriticalException; import org.springframework.batch.item.ExecutionAttributes; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemRecoverer; import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.KeyedItemReader; import org.springframework.batch.item.exception.ResetFailedException; import org.springframework.batch.item.stream.SimpleStreamManager; import org.springframework.batch.item.stream.StreamManager; @@ -48,30 +51,31 @@ import org.springframework.batch.repeat.exception.handler.ExceptionHandler; import org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHandler; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.batch.repeat.support.RepeatTemplate; +import org.springframework.batch.retry.RetryPolicy; +import org.springframework.batch.retry.callback.ItemReaderRetryCallback; +import org.springframework.batch.retry.policy.ItemReaderRetryPolicy; +import org.springframework.batch.retry.support.RetryTemplate; +import org.springframework.beans.factory.InitializingBean; import org.springframework.transaction.TransactionStatus; import org.springframework.util.Assert; /** - * Simple implementation of executing the step as a set of chunks, each chunk - * surrounded by a transaction. The structure is therefore that of two nested - * loops, with transaction boundary around the whole inner loop. The outer loop - * is controlled by the step operations ({@link #setStepOperations(RepeatOperations)}), - * and the inner loop by the chunk operations ({@link #setChunkOperations(RepeatOperations)}). - * The inner loop should always be executed in a single thread, so the chunk - * operations should not do any concurrent execution. N.B. usually that means - * that the chunk operations should be a {@link RepeatTemplate} (which is the - * default).
+ * Simple implementation of executing the step as a set of chunks, each chunk surrounded by a transaction. The structure + * is therefore that of two nested loops, with transaction boundary around the whole inner loop. The outer loop is + * controlled by the step operations ({@link #setStepOperations(RepeatOperations)}), and the inner loop by the chunk + * operations ({@link #setChunkOperations(RepeatOperations)}). The inner loop should always be executed in a single + * thread, so the chunk operations should not do any concurrent execution. N.B. usually that means that the chunk + * operations should be a {@link RepeatTemplate} (which is the default).
* - * Clients can use interceptors in the step operations to intercept or listen to - * the iteration on a step-wide basis, for instance to get a callback when the - * step is complete. Those that want callbacks at the level of an individual - * tasks, can specify interceptors for the chunk operations. + * Clients can use interceptors in the step operations to intercept or listen to the iteration on a step-wide basis, for + * instance to get a callback when the step is complete. Those that want callbacks at the level of an individual tasks, + * can specify interceptors for the chunk operations. * * @author Dave Syer * @author Lucas Ward - * + * @author Ben Hale */ -public class SimpleStepExecutor { +public class SimpleStepExecutor implements InitializingBean { private static final Log logger = LogFactory.getLog(SimpleStepExecutor.class); @@ -86,12 +90,22 @@ public class SimpleStepExecutor { // default to checking current thread for interruption. private StepInterruptionPolicy interruptionPolicy = new ThreadStepInterruptionPolicy(); - private Tasklet tasklet; - private AbstractStep step; private StreamManager streamManager; + private ItemReader itemReader; + + private ItemWriter itemWriter; + + private RetryPolicy retryPolicy = null; + + private ItemRecoverer itemRecoverer; + + private RetryTemplate template = new RetryTemplate(); + + private ItemReaderRetryCallback retryCallback; + /** * Package private constructor so the step can create a the executor. */ @@ -100,22 +114,18 @@ public class SimpleStepExecutor { } /** - * Public setter for the {@link StreamManager}. This will be used to create - * the {@link StepContext}, and hence any component that is a - * {@link ItemStream} and in step scope will be registered with the service. - * The {@link StepContext} is then a source of aggregate statistics for the - * step. + * Public setter for the {@link StreamManager}. This will be used to create the {@link StepContext}, and hence any + * component that is a {@link ItemStream} and in step scope will be registered with the service. The + * {@link StepContext} is then a source of aggregate statistics for the step. * - * @param streamManager the {@link StreamManager} to set. Default is a - * {@link SimpleStreamManager}. + * @param streamManager the {@link StreamManager} to set. Default is a {@link SimpleStreamManager}. */ public void setStreamManager(StreamManager streamManager) { this.streamManager = streamManager; } /** - * Injected strategy for storage and retrieval of persistent step - * information. Mandatory property. + * Injected strategy for storage and retrieval of persistent step information. Mandatory property. * * @param jobRepository */ @@ -124,9 +134,8 @@ public class SimpleStepExecutor { } /** - * The {@link RepeatOperations} to use for the outer loop of the batch - * processing. Should be set up by the caller through a factory. Defaults to - * a plain {@link RepeatTemplate}. + * The {@link RepeatOperations} to use for the outer loop of the batch processing. Should be set up by the caller + * through a factory. Defaults to a plain {@link RepeatTemplate}. * * @param stepOperations a {@link RepeatOperations} instance. */ @@ -135,9 +144,8 @@ public class SimpleStepExecutor { } /** - * The {@link RepeatOperations} to use for the inner loop of the batch - * processing. Should be set up by the caller through a factory. Defaults to - * a plain {@link RepeatTemplate}. + * The {@link RepeatOperations} to use for the inner loop of the batch processing. Should be set up by the caller + * through a factory. Defaults to a plain {@link RepeatTemplate}. * * @param chunkOperations a {@link RepeatOperations} instance. */ @@ -146,18 +154,121 @@ public class SimpleStepExecutor { } /** - * Process the step and update its context so that progress can be monitored - * by the caller. The step is broken down into chunks, each one executing in - * a transaction. The step and its execution and execution context are all - * given an up to date {@link BatchStatus}, and the {@link JobRepository} - * is used to store the result. Various reporting information are also added - * to the current context (the {@link RepeatContext} governing the step - * execution, which would normally be available to the caller somehow - * through the step's {@link JobExecutionContext}.
+ * Setter for the {@link StepInterruptionPolicy}. The policy is used to check whether an external request has been + * made to interrupt the job execution. + * + * @param interruptionPolicy a {@link StepInterruptionPolicy} + */ + public void setInterruptionPolicy(StepInterruptionPolicy interruptionPolicy) { + this.interruptionPolicy = interruptionPolicy; + } + + /** + * Setter for the {@link ExitCodeExceptionClassifier} that will be used to classify any exception that causes a job + * to fail. + * + * @param exceptionClassifier + */ + public void setExceptionClassifier(ExitCodeExceptionClassifier exceptionClassifier) { + this.exceptionClassifier = exceptionClassifier; + } + + /** + * @param itemReader + */ + public void setItemReader(ItemReader itemReader) { + this.itemReader = itemReader; + } + + /** + * @param itemWriter + */ + public void setItemWriter(ItemWriter itemWriter) { + this.itemWriter = itemWriter; + } + + /** + * Setter for injecting optional recovery handler. + * + * @param itemRecoverer + */ + public void setItemRecoverer(ItemRecoverer itemRecoverer) { + this.itemRecoverer = itemRecoverer; + } + + /** + * Public setter for the retryPolicy. + * + * @param retyPolicy the retryPolicy to set + */ + public void setRetryPolicy(RetryPolicy retryPolicy) { + this.retryPolicy = retryPolicy; + } + + /** + * Check mandatory properties (reader and writer). + * + * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() + */ + public void afterPropertiesSet() throws Exception { + Assert.notNull(itemReader, "ItemReader must be provided"); + Assert.notNull(itemWriter, "ItemWriter must be provided"); + + if (itemRecoverer == null && (itemReader instanceof ItemRecoverer)) { + itemRecoverer = (ItemRecoverer) itemReader; + } + + ItemReaderRetryPolicy itemProviderRetryPolicy = new ItemReaderRetryPolicy(retryPolicy); + template.setRetryPolicy(itemProviderRetryPolicy); + + if (retryPolicy != null) { + Assert.state(itemReader instanceof KeyedItemReader, + "ItemReader must be instance of KeyedItemReader to use the retry policy"); + retryCallback = new ItemReaderRetryCallback((KeyedItemReader) itemReader, itemWriter); + retryCallback.setRecoverer(itemRecoverer); + } + + } + + /** + * Apply the configuration by inspecting it to see if it has any relevant policy information. + * + * @param step a step + */ + void applyConfiguration(AbstractStep step) { + + if (step instanceof SimpleStep) { + SimpleStep simple = (SimpleStep) step; + if (this.chunkOperations instanceof RepeatTemplate) { + RepeatTemplate template = (RepeatTemplate) this.chunkOperations; + template.setCompletionPolicy(new SimpleCompletionPolicy(simple.getCommitInterval())); + } + } + + ExceptionHandler exceptionHandler = step.getExceptionHandler(); + + if (step.getSkipLimit() > 0 && exceptionHandler == null) { + SimpleLimitExceptionHandler handler = new SimpleLimitExceptionHandler(); + handler.setLimit(step.getSkipLimit()); + exceptionHandler = handler; + } + + if (this.stepOperations instanceof RepeatTemplate && exceptionHandler != null) { + RepeatTemplate template = (RepeatTemplate) this.stepOperations; + template.setExceptionHandler(exceptionHandler); + } + + } + + /** + * Process the step and update its context so that progress can be monitored by the caller. The step is broken down + * into chunks, each one executing in a transaction. The step and its execution and execution context are all given + * an up to date {@link BatchStatus}, and the {@link JobRepository} is used to store the result. Various reporting + * information are also added to the current context (the {@link RepeatContext} governing the step execution, which + * would normally be available to the caller somehow through the step's {@link JobExecutionContext}.
* * @throws StepInterruptedException if the step or a chunk is interrupted - * @throws RuntimeException if there is an exception during a chunk - * execution + * @throws RuntimeException if there is an exception during a chunk execution * @see StepExecutor#execute(StepExecution) */ public void execute(final StepExecution stepExecution) throws BatchCriticalException, StepInterruptedException { @@ -205,8 +316,7 @@ public class SimpleStepExecutor { try { /* - * New transaction obtained, resynchronize - * TransactionSynchronization objects + * New transaction obtained, resynchronize TransactionSynchronization objects */ result = processChunk(step, contribution); @@ -235,38 +345,31 @@ public class SimpleStepExecutor { streamManager.commit(transaction); - } - catch (Throwable t) { + } catch (Throwable t) { /* - * Any exception thrown within the transaction template - * will automatically cause the transaction to rollback. - * We need to include exceptions during an attempted - * commit (e.g. Hibernate flush) so this catch block - * comes outside the transaction. + * Any exception thrown within the transaction template will automatically cause the transaction + * to rollback. We need to include exceptions during an attempted commit (e.g. Hibernate flush) + * so this catch block comes outside the transaction. */ synchronized (stepExecution) { stepExecution.rollback(); } try { streamManager.rollback(transaction); - } - catch (ResetFailedException e) { + } catch (ResetFailedException e) { // The original Throwable cause is in danger of // being lost here, so we log the reset // failure and re-throw with cause of the rollback. logger.error("Encountered reset error on rollback: " - + "one of the streams may be in an inconsistent state, " - + "so this step should not proceed", e); - throw new ResetFailedException( - "Encountered reset error on rollback. " + - "Consult logs for the cause of the reet failure. " + - "The cause of the original rollback is incuded here.", - t); + + "one of the streams may be in an inconsistent state, " + + "so this step should not proceed", e); + throw new ResetFailedException("Encountered reset error on rollback. " + + "Consult logs for the cause of the reet failure. " + + "The cause of the original rollback is incuded here.", t); } if (t instanceof RuntimeException) { throw (RuntimeException) t; - } - else { + } else { throw new RuntimeException(t); } } @@ -282,32 +385,27 @@ public class SimpleStepExecutor { }); updateStatus(stepExecution, BatchStatus.COMPLETED); - } - catch (RuntimeException e) { + } catch (RuntimeException e) { // classify exception so an exit code can be stored. status = exceptionClassifier.classifyForExitCode(e); if (e.getCause() instanceof StepInterruptedException) { updateStatus(stepExecution, BatchStatus.STOPPED); throw (StepInterruptedException) e.getCause(); - } - else if (e instanceof ResetFailedException) { + } else if (e instanceof ResetFailedException) { updateStatus(stepExecution, BatchStatus.UNKNOWN); throw (ResetFailedException) e; - } - else { + } else { updateStatus(stepExecution, BatchStatus.FAILED); throw e; } - } - finally { + } finally { stepExecution.setExitStatus(status); stepExecution.setEndTime(new Date(System.currentTimeMillis())); try { jobRepository.saveOrUpdate(stepExecution); - } - finally { + } finally { // clear any registered synchronizations StepSynchronizationManager.close(); } @@ -315,6 +413,116 @@ public class SimpleStepExecutor { } + /** + * Execute a bunch of identical business logic operations all within a transaction. The transaction is + * programmatically started and stopped outside this method, so subclasses that override do not need to create a + * transaction. + * + * @param step the current step containing the {@link Tasklet} with the business logic. + * @return true if there is more data to process. + */ + ExitStatus processChunk(final Step step, final StepContribution contribution) { + ExitStatus result = chunkOperations.iterate(new RepeatCallback() { + public ExitStatus doInIteration(final RepeatContext context) throws Exception { + if (contribution.isTerminateOnly()) { + context.setTerminateOnly(); + } + // check for interruption before each item as well + interruptionPolicy.checkInterrupted(context); + ExitStatus exitStatus = doProcessing(contribution); + contribution.incrementTaskCount(); + // check for interruption after each item as well + interruptionPolicy.checkInterrupted(context); + return exitStatus; + } + }); + return result; + } + + /** + * Execute the business logic, delegating to the given {@link Tasklet}. Subclasses could extend the behaviour as + * long as they always return the value of this method call in their superclass.
+ * + * If there is an exception and the {@link Tasklet} implements {@link Skippable} then the skip method is called. + * + * @param tasklet the unit of business logic to execute + * @param contribution the current step + * @return boolean if there is more processing to do + * @throws Exception if there is an error + */ + private ExitStatus doProcessing(StepContribution contribution) throws Exception { + ExitStatus exitStatus = ExitStatus.CONTINUABLE; + + try { + + exitStatus = execute(); + + } catch (Exception e) { + skip(); + // Rethrow so that outer transaction is rolled back properly + throw e; + + } + + return exitStatus; + } + + /** + * Read from the {@link ItemReader} and process (if not null) with the {@link ItemWriter}. The call to + * {@link ItemWriter} is wrapped in a stateful retry, if a {@link RetryPolicy} is provided. The + * {@link ItemRecoverer} is used (if provided) in the case of an exception to apply alternate processing to the + * item. If the stateful retry is in place then the recovery will happen in the next transaction automatically, + * otherwise it might be necessary for clients to make the recover method transactional with appropriate propagation + * behaviour (probably REQUIRES_NEW because the call will happen in the context of a transaction that is about to + * rollback). + * + * @see org.springframework.batch.core.tasklet.Tasklet#execute() + */ + private ExitStatus execute() throws Exception { + + if (retryCallback == null) { + Object item = itemReader.read(); + if (item == null) { + return ExitStatus.FINISHED; + } + try { + itemWriter.write(item); + } catch (Exception e) { + if (itemRecoverer != null) { + itemRecoverer.recover(item, e); + } + // Re-throw the exception so that the surrounding transaction + // rolls back if there is one + throw e; + } + return ExitStatus.CONTINUABLE; + } + + return new ExitStatus(template.execute(retryCallback) != null); + + } + + /** + * Mark the current item as skipped if possible. If there is a retry policy in action there is no need to take any + * action now because it will be covered by the retry in the next transaction. Otherwise if the reader and / or + * writer are {@link Skippable} then delegate to them in that order. + * + * @see org.springframework.batch.io.Skippable#skip() + */ + private void skip() { + if (retryCallback != null) { + // No need to skip because the recoverer will take any action + // necessary. + return; + } + if (this.itemReader instanceof Skippable) { + ((Skippable) this.itemReader).skip(); + } + if (this.itemWriter instanceof Skippable) { + ((Skippable) this.itemWriter).skip(); + } + } + /** * Convenience method to update the status in all relevant places. * @@ -328,126 +536,4 @@ public class SimpleStepExecutor { jobRepository.update(step); jobRepository.saveOrUpdate(stepExecution); } - - /** - * Execute a bunch of identical business logic operations all within a - * transaction. The transaction is programmatically started and stopped - * outside this method, so subclasses that override do not need to create a - * transaction. - * - * @param step the current step containing the {@link Tasklet} with the - * business logic. - * @return true if there is more data to process. - */ - protected final ExitStatus processChunk(final Step step, final StepContribution contribution) { - ExitStatus result = chunkOperations.iterate(new RepeatCallback() { - public ExitStatus doInIteration(final RepeatContext context) throws Exception { - if (contribution.isTerminateOnly()) { - context.setTerminateOnly(); - } - // check for interruption before each item as well - interruptionPolicy.checkInterrupted(context); - ExitStatus exitStatus = doTaskletProcessing(tasklet, contribution); - contribution.incrementTaskCount(); - // check for interruption after each item as well - interruptionPolicy.checkInterrupted(context); - return exitStatus; - } - }); - return result; - } - - /** - * Execute the business logic, delegating to the given {@link Tasklet}. - * Subclasses could extend the behaviour as long as they always return the - * value of this method call in their superclass.
- * - * If there is an exception and the {@link Tasklet} implements - * {@link Skippable} then the skip method is called. - * - * @param tasklet the unit of business logic to execute - * @param contribution the current step - * @return boolean if there is more processing to do - * @throws Exception if there is an error - */ - protected ExitStatus doTaskletProcessing(Tasklet tasklet, StepContribution contribution) throws Exception { - ExitStatus exitStatus = ExitStatus.CONTINUABLE; - - try { - - exitStatus = tasklet.execute(); - - } - catch (Exception e) { - - if (tasklet instanceof Skippable) { - ((Skippable) tasklet).skip(); - } - - // Rethrow so that outer transaction is rolled back properly - throw e; - - } - - return exitStatus; - } - - /** - * Setter for the {@link StepInterruptionPolicy}. The policy is used to - * check whether an external request has been made to interrupt the job - * execution. - * - * @param interruptionPolicy a {@link StepInterruptionPolicy} - */ - public void setInterruptionPolicy(StepInterruptionPolicy interruptionPolicy) { - this.interruptionPolicy = interruptionPolicy; - } - - /** - * Setter for the {@link ExitCodeExceptionClassifier} that will be used to - * classify any exception that causes a job to fail. - * - * @param exceptionClassifier - */ - public void setExceptionClassifier(ExitCodeExceptionClassifier exceptionClassifier) { - this.exceptionClassifier = exceptionClassifier; - } - - /** - * Apply the configuration by inspecting it to see if it has any relevant - * policy information. - * - * @param step a step - */ - void applyConfiguration(AbstractStep step) { - - if (step instanceof SimpleStep) { - SimpleStep simple = (SimpleStep) step; - if (this.chunkOperations instanceof RepeatTemplate) { - RepeatTemplate template = (RepeatTemplate) this.chunkOperations; - template.setCompletionPolicy(new SimpleCompletionPolicy(simple.getCommitInterval())); - } - } - - ExceptionHandler exceptionHandler = step.getExceptionHandler(); - - if (step.getSkipLimit() > 0 && exceptionHandler == null) { - SimpleLimitExceptionHandler handler = new SimpleLimitExceptionHandler(); - handler.setLimit(step.getSkipLimit()); - exceptionHandler = handler; - } - - if (this.stepOperations instanceof RepeatTemplate && exceptionHandler != null) { - RepeatTemplate template = (RepeatTemplate) this.stepOperations; - template.setExceptionHandler(exceptionHandler); - } - - } - - /** - * @param tasklet a {@link Tasklet} to execute when a step is processed - */ - public void setTasklet(Tasklet tasklet) { - this.tasklet = tasklet; - } } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/tasklet/ItemOrientedTasklet.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/tasklet/ItemOrientedTasklet.java deleted file mode 100644 index f254cf5f4..000000000 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/tasklet/ItemOrientedTasklet.java +++ /dev/null @@ -1,215 +0,0 @@ -/* - * Copyright 2006-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.batch.execution.tasklet; - -import org.springframework.batch.core.tasklet.Tasklet; -import org.springframework.batch.io.Skippable; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemRecoverer; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.KeyedItemReader; -import org.springframework.batch.repeat.ExitStatus; -import org.springframework.batch.retry.RetryPolicy; -import org.springframework.batch.retry.callback.ItemReaderRetryCallback; -import org.springframework.batch.retry.policy.ItemReaderRetryPolicy; -import org.springframework.batch.retry.support.RetryTemplate; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.util.Assert; - -/** - * A concrete implementation of the {@link Tasklet} interface that provides - * 'split processing'. This type of processing is characterized by separating - * the reading and processing of batch data into two separate classes: - * {@link ItemReader} and {@link ItemWriter}. The {@link ItemReader} class - * provides a solid means for re-usability and enforces good architecture - * practices. Because an object must be returned by the - * {@link ItemReader} to continue processing, (returning null indicates - * processing should end) a developer is forced to read in all relevant data, - * place it into a domain object, and return that object. The - * {@link ItemWriter} will then use this object for calculations and output.
- * - * If a {@link RetryPolicy} is provided it will be used to construct a stateful - * retry around the {@link ItemWriter}, delegating identity concerns to the - * {@link ItemReader} and recovery concerns to the {@link ItemRecoverer} (if - * present). In this case clients of this class do not need to take any - * additional action at runtime to take advantage of the retry and recovery, - * provided that when the {@link #execute()} method is called again the same - * item is eventually re-presented (normally this would be the case because a - * transaction would have rolled back and the {@link ItemReader} would go back - * to its previous state).
- * - * If a {@link RetryPolicy} is not provided then the {@link ItemRecoverer} can - * be used to attempt to recover immediately (with no retry) from a processing - * error. Clients of this class should ensure that the recovery takes place in a - * separate transaction (e.g. with propagation REQUIRES_NEW) if necessary. This - * can be achieved by injecting an {@link ItemRecoverer} that has a - * transactional recover method. - * - * @see ItemReader - * @see ItemWriter - * @see RetryPolicy - * @see Recoverable - * - * @author Lucas Ward - * @author Dave Syer - * @author Robert Kasanicky - * - */ -public class ItemOrientedTasklet implements Tasklet, Skippable, InitializingBean { - - /** - * Prefix added to statistics keys from writer if needed to avoid - * ambiguity between reader and writer. - */ - public static final String WRITER_STATISTICS_PREFIX = "writer."; - - /** - * Prefix added to statistics keys from reader if needed to avoid - * ambiguity between provider and writer. - */ - public static final String READER_STATISTICS_PREFIX = "reader."; - - private RetryPolicy retryPolicy = null; - - protected ItemReader itemReader; - - protected ItemWriter itemWriter; - - private ItemRecoverer itemRecoverer; - - private RetryTemplate template = new RetryTemplate(); - - private ItemReaderRetryCallback retryCallback; - - /** - * Check mandatory properties (reader and writer). - * - * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() - */ - public void afterPropertiesSet() throws Exception { - Assert.notNull(itemReader, "ItemReader must be provided"); - Assert.notNull(itemWriter, "ItemWriter must be provided"); - - if (itemRecoverer == null && (itemReader instanceof ItemRecoverer)) { - itemRecoverer = (ItemRecoverer) itemReader; - } - - ItemReaderRetryPolicy itemProviderRetryPolicy = new ItemReaderRetryPolicy(retryPolicy); - template.setRetryPolicy(itemProviderRetryPolicy); - - if (retryPolicy != null) { - Assert.state(itemReader instanceof KeyedItemReader, "ItemReader must be instance of KeyedItemReader to use the retry policy"); - retryCallback = new ItemReaderRetryCallback((KeyedItemReader) itemReader, itemWriter); - retryCallback.setRecoverer(itemRecoverer); - } - - } - - /** - * Read from the {@link ItemReader} and process (if not null) with the - * {@link ItemWriter}. The call to {@link ItemWriter} is wrapped in a - * stateful retry, if a {@link RetryPolicy} is provided. The - * {@link ItemRecoverer} is used (if provided) in the case of an exception - * to apply alternate processing to the item. If the stateful retry is in - * place then the recovery will happen in the next transaction - * automatically, otherwise it might be necessary for clients to make the - * recover method transactional with appropriate propagation behaviour - * (probably REQUIRES_NEW because the call will happen in the context of a - * transaction that is about to rollback). - * - * @see org.springframework.batch.core.tasklet.Tasklet#execute() - */ - public ExitStatus execute() throws Exception { - - if (retryCallback == null) { - Object item = itemReader.read(); - if (item == null) { - return ExitStatus.FINISHED; - } - try { - itemWriter.write(item); - } - catch (Exception e) { - if (itemRecoverer != null) { - itemRecoverer.recover(item, e); - } - // Re-throw the exception so that the surrounding transaction - // rolls back if there is one - throw e; - } - return ExitStatus.CONTINUABLE; - } - - return new ExitStatus(template.execute(retryCallback) != null); - - } - - /** - * @param itemReader - */ - public void setItemReader(ItemReader itemReader) { - this.itemReader = itemReader; - } - - /** - * @param itemWriter - */ - public void setItemWriter(ItemWriter itemWriter) { - this.itemWriter = itemWriter; - } - - /** - * Setter for injecting optional recovery handler. - * - * @param itemRecoverer - */ - public void setItemRecoverer(ItemRecoverer itemRecoverer) { - this.itemRecoverer = itemRecoverer; - } - - /** - * Mark the current item as skipped if possible. If there is a retry policy - * in action there is no need to take any action now because it will be - * covered by the retry in the next transaction. Otherwise if the reader - * and / or writer are {@link Skippable} then delegate to them in that - * order. - * - * @see org.springframework.batch.io.Skippable#skip() - */ - public void skip() { - if (retryCallback != null) { - // No need to skip because the recoverer will take any action - // necessary. - return; - } - if (this.itemReader instanceof Skippable) { - ((Skippable) this.itemReader).skip(); - } - if (this.itemWriter instanceof Skippable) { - ((Skippable) this.itemWriter).skip(); - } - } - - /** - * Public setter for the retryPolicy. - * - * @param retyPolicy the retryPolicy to set - */ - public void setRetryPolicy(RetryPolicy retryPolicy) { - this.retryPolicy = retryPolicy; - } -} diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/job/simple/SimpleJobTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/job/simple/SimpleJobTests.java index 6f26e7f5d..3cfea6112 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/job/simple/SimpleJobTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/job/simple/SimpleJobTests.java @@ -251,6 +251,7 @@ public class SimpleJobTests extends TestCase { private Runnable runnable; private Exception exception; + private Tasklet tasklet; /** * @param string @@ -265,6 +266,13 @@ public class SimpleJobTests extends TestCase { public void setProcessException(Exception exception) { this.exception = exception; } + + /** + * @param tasklet the tasklet to set + */ + public void setTasklet(Tasklet tasklet) { + this.tasklet = tasklet; + } /** * @param runnable diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java index c318bf717..4c244e677 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java @@ -27,15 +27,12 @@ import org.springframework.batch.core.domain.BatchStatus; import org.springframework.batch.core.domain.JobExecution; import org.springframework.batch.core.domain.JobInstance; import org.springframework.batch.core.domain.JobParameters; -import org.springframework.batch.core.tasklet.Tasklet; import org.springframework.batch.execution.job.simple.SimpleJob; import org.springframework.batch.execution.repository.SimpleJobRepository; import org.springframework.batch.execution.repository.dao.MapJobDao; import org.springframework.batch.execution.repository.dao.MapStepDao; import org.springframework.batch.execution.step.simple.AbstractStep; -import org.springframework.batch.execution.step.simple.RepeatOperationsStep; import org.springframework.batch.execution.step.simple.SimpleStep; -import org.springframework.batch.execution.tasklet.ItemOrientedTasklet; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemRecoverer; import org.springframework.batch.item.ItemWriter; @@ -71,42 +68,40 @@ public class SimpleJobTests extends TestCase { job.setJobRepository(repository); } - private Tasklet getTasklet(String arg) throws Exception { - return getTasklet(new String[] { arg }); + private AbstractStep getStep(String arg) throws Exception { + return getStep(new String[] { arg }); } - private Tasklet getTasklet(String arg0, String arg1) throws Exception { - return getTasklet(new String[] { arg0, arg1 }); + private AbstractStep getStep(String arg0, String arg1) throws Exception { + return getStep(new String[] { arg0, arg1 }); } - - private ItemOrientedTasklet getTasklet(String[] args) throws Exception { - ItemOrientedTasklet module = new ItemOrientedTasklet(); + + private AbstractStep getStep(String[] args) throws Exception { + SimpleStep step = new SimpleStep(); List items = TransactionAwareProxyFactory.createTransactionalList(); items.addAll(Arrays.asList(args)); provider = new ListItemReader(items); - module.setItemRecoverer(new ItemRecoverer() { + step.setItemRecoverer(new ItemRecoverer() { public boolean recover(Object item, Throwable cause) { recovered.add(item); assertTrue(TransactionSynchronizationManager.isActualTransactionActive()); return true; } }); - module.setItemReader(provider); - module.setItemWriter(processor); - module.afterPropertiesSet(); - return module; + step.setItemReader(provider); + step.setItemWriter(processor); + step.setJobRepository(repository); + step.setTransactionManager(new ResourcelessTransactionManager()); + step.afterPropertiesSet(); + return step; } public void testSimpleJob() throws Exception { job.setSteps(new ArrayList()); - AbstractStep step = new SimpleStep(getTasklet("foo", "bar")); - step.setJobRepository(repository); - step.setTransactionManager(new ResourcelessTransactionManager()); + AbstractStep step = getStep("foo", "bar"); job.addStep(step); - step = new SimpleStep(getTasklet("spam")); - step.setJobRepository(repository); - step.setTransactionManager(new ResourcelessTransactionManager()); + step = getStep("spam"); job.addStep(step); JobInstance jobInstance = repository.createJobExecution(job, new JobParameters()).getJobInstance(); @@ -137,18 +132,19 @@ public class SimpleJobTests extends TestCase { * is recovered ("skipped") on the second attempt (see retry policy * definition above)... */ - final ItemOrientedTasklet module = getTasklet(new String[] { "foo", "bar", "spam" }); - RepeatOperationsStep step = new RepeatOperationsStep(); - step.setTasklet(module); - step.setChunkOperations(chunkOperations); - step.setJobRepository(repository); - step.setTransactionManager(new ResourcelessTransactionManager()); - module.setItemWriter(new AbstractItemWriter() { + AbstractStep step = getStep(new String[] { "foo", "bar", "spam" }); + + +// Tasklet module = getTasklet(new String[] { "foo", "bar", "spam" }); +// RepeatOperationsStep step = new RepeatOperationsStep(); +// step.setTasklet(module); +// step.setChunkOperations(chunkOperations); + step.setItemWriter(new AbstractItemWriter() { public void write(Object data) throws Exception { throw new RuntimeException("Try again Dummy!"); } }); - module.afterPropertiesSet(); + step.afterPropertiesSet(); job.setSteps(Collections.singletonList(step)); JobExecution jobExecution = repository.createJobExecution(job, new JobParameters()); @@ -162,17 +158,14 @@ public class SimpleJobTests extends TestCase { } public void testExceptionTerminates() throws Exception { - - final ItemOrientedTasklet module = getTasklet(new String[] { "foo", "bar", "spam" }); - AbstractStep step = new SimpleStep(module); - step.setJobRepository(repository); - step.setTransactionManager(new ResourcelessTransactionManager()); - module.setItemWriter(new AbstractItemWriter() { +// Tasklet module = getTasklet(new String[] { "foo", "bar", "spam" }); + AbstractStep step = getStep(new String[] { "foo", "bar", "spam" }); + step.setItemWriter(new AbstractItemWriter() { public void write(Object data) throws Exception { throw new RuntimeException("Foo"); } }); - module.afterPropertiesSet(); + step.afterPropertiesSet(); job.setSteps(Collections.singletonList(step)); JobExecution jobExecution = repository.createJobExecution(job, new JobParameters()); diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepConfigurationTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepConfigurationTests.java index b188dca71..eac86b34e 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepConfigurationTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepConfigurationTests.java @@ -18,6 +18,8 @@ package org.springframework.batch.execution.step.simple; import junit.framework.TestCase; import org.springframework.batch.core.tasklet.Tasklet; +import org.springframework.batch.item.reader.AbstractItemReader; +import org.springframework.batch.item.writer.AbstractItemWriter; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.repeat.exception.handler.DefaultExceptionHandler; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; @@ -31,8 +33,7 @@ public class SimpleStepConfigurationTests extends TestCase { SimpleStep configuration = new SimpleStep("foo"); /** - * Test method for - * {@link org.springframework.batch.execution.step.simple.SimpleStep#SimpleStepConfiguration()}. + * Test method for {@link org.springframework.batch.execution.step.simple.SimpleStep#SimpleStepConfiguration()}. */ public void testSimpleStepConfiguration() { assertNotNull(configuration.getName()); @@ -43,23 +44,32 @@ public class SimpleStepConfigurationTests extends TestCase { /** * Test method for * {@link org.springframework.batch.execution.step.simple.SimpleStep#SimpleStepConfiguration(org.springframework.batch.core.tasklet.Tasklet)}. + * * @throws Exception */ public void testSimpleStepConfigurationTasklet() throws Exception { - Tasklet tasklet = new Tasklet() { - public ExitStatus execute() throws Exception { - return ExitStatus.FINISHED; + configuration = new SimpleStep(); + configuration.setItemReader(new AbstractItemReader() { + + public Object read() throws Exception { + // TODO Auto-generated method stub + return null; } - }; - configuration = new SimpleStep(tasklet); + }); + configuration.setItemWriter(new AbstractItemWriter() { + + public void write(Object item) throws Exception { + // TODO Auto-generated method stub + + } + }); configuration.setJobRepository(new JobRepositorySupport()); configuration.setTransactionManager(new ResourcelessTransactionManager()); configuration.afterPropertiesSet(); } /** - * Test method for - * {@link org.springframework.batch.execution.step.simple.SimpleStep#getCommitInterval()}. + * Test method for {@link org.springframework.batch.execution.step.simple.SimpleStep#getCommitInterval()}. */ public void testGetCommitInterval() { assertEquals(1, configuration.getCommitInterval()); @@ -68,8 +78,7 @@ public class SimpleStepConfigurationTests extends TestCase { } /** - * Test method for - * {@link org.springframework.batch.execution.step.simple.AbstractStep#getExceptionHandler()}. + * Test method for {@link org.springframework.batch.execution.step.simple.AbstractStep#getExceptionHandler()}. */ public void testGetExceptionHandler() { assertNull(configuration.getExceptionHandler()); @@ -78,8 +87,7 @@ public class SimpleStepConfigurationTests extends TestCase { } /** - * Test method for - * {@link org.springframework.batch.execution.step.simple.AbstractStep#getExceptionHandler()}. + * Test method for {@link org.springframework.batch.execution.step.simple.AbstractStep#getExceptionHandler()}. */ public void testSkipLimit() { assertEquals(0, configuration.getSkipLimit()); @@ -88,8 +96,7 @@ public class SimpleStepConfigurationTests extends TestCase { } /** - * Test method for - * {@link org.springframework.batch.execution.step.simple.AbstractStep#getSkipLimit()}. + * Test method for {@link org.springframework.batch.execution.step.simple.AbstractStep#getSkipLimit()}. */ public void testGetSkipLimit() { assertEquals(0, configuration.getSkipLimit()); @@ -98,8 +105,7 @@ public class SimpleStepConfigurationTests extends TestCase { } /** - * Test method for - * {@link org.springframework.batch.execution.step.simple.AbstractStep#isSaveExecutionAttributes()}. + * Test method for {@link org.springframework.batch.execution.step.simple.AbstractStep#isSaveExecutionAttributes()}. */ public void testIsSaveExecutionAttributes() { assertEquals(false, configuration.isSaveExecutionAttributes()); diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java index f10b8ca76..f63820119 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java @@ -29,6 +29,7 @@ import org.springframework.batch.core.domain.JobExecution; import org.springframework.batch.core.domain.JobInstance; import org.springframework.batch.core.domain.JobParameters; import org.springframework.batch.core.domain.JobSupport; +import org.springframework.batch.core.domain.Step; import org.springframework.batch.core.domain.StepContribution; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.core.domain.StepInstance; @@ -39,7 +40,6 @@ import org.springframework.batch.execution.repository.dao.MapJobDao; import org.springframework.batch.execution.repository.dao.MapStepDao; import org.springframework.batch.execution.scope.StepScope; import org.springframework.batch.execution.scope.StepSynchronizationManager; -import org.springframework.batch.execution.tasklet.ItemOrientedTasklet; import org.springframework.batch.item.ExecutionAttributes; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; @@ -84,31 +84,21 @@ public class SimpleStepExecutorTests extends TestCase { return new ListItemReader(Arrays.asList(args)); } - /** - * @param strings - * @return - * @throws Exception - */ - private Tasklet getTasklet(String[] strings) throws Exception { - ItemOrientedTasklet module = new ItemOrientedTasklet(); - module.setItemWriter(processor); - module.setItemReader(getReader(strings)); - module.afterPropertiesSet(); - return module; + + + private AbstractStep getStep(String[] strings) throws Exception { + SimpleStep step = new SimpleStep(); + step.setItemWriter(processor); + step.setItemReader(getReader(strings)); + step.setJobRepository(new JobRepositorySupport()); + step.setTransactionManager(transactionManager); + step.afterPropertiesSet(); + return step; } - /* - * (non-Javadoc) - * - * @see junit.framework.TestCase#setUp() - */ protected void setUp() throws Exception { - super.setUp(); transactionManager = new ResourcelessTransactionManager(); - stepConfiguration = new SimpleStep(); - stepConfiguration.setTasklet(getTasklet(new String[] { "foo", "bar", "spam" })); - stepConfiguration.setJobRepository(new JobRepositorySupport()); - stepConfiguration.setTransactionManager(transactionManager); + stepConfiguration = getStep(new String[] { "foo", "bar", "spam" }); stepExecutor = (SimpleStepExecutor) stepConfiguration.createStepExecutor(); template = new RepeatTemplate(); diff --git a/spring-batch-execution/src/test/resources/simple-container-definition.xml b/spring-batch-execution/src/test/resources/simple-container-definition.xml index 27cdaa6a3..aef82fe5c 100644 --- a/spring-batch-execution/src/test/resources/simple-container-definition.xml +++ b/spring-batch-execution/src/test/resources/simple-container-definition.xml @@ -18,7 +18,8 @@ class="org.springframework.batch.execution.configuration.MapJobRegistry" /> + class="org.springframework.batch.execution.job.simple.SimpleJob" + abstract="true"> @@ -35,8 +36,8 @@ - - + +