diff --git a/execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java b/execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java index 3c86473eb..f361797a5 100644 --- a/execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java +++ b/execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java @@ -59,24 +59,18 @@ import org.springframework.util.Assert; * 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. - * + * * @author Dave Syer * @author Lucas Ward - * + * */ public class SimpleStepExecutor implements StepExecutor { - /** - * Key placed in step scope context to identify the - * {@link StepExecution}. - */ - public static final String STEP_KEY = "STEP"; - /** * Context attribute key for step execution. Used by monitoring and managing * clients to inspect current step execution. @@ -102,13 +96,15 @@ public class SimpleStepExecutor implements StepExecutor { // Not for production use... protected PlatformTransactionManager transactionManager = new ResourcelessTransactionManager(); - public void setTransactionManager(PlatformTransactionManager transactionManager) { + public void setTransactionManager( + PlatformTransactionManager transactionManager) { this.transactionManager = transactionManager; } /** * Injected strategy for storage and retrieval of persistent step * information. Mandatory property. + * * @param jobRepository */ public void setRepository(JobRepository jobRepository) { @@ -119,7 +115,9 @@ public class SimpleStepExecutor implements StepExecutor { * 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. + * + * @param stepOperations + * a {@link RepeatOperations} instance. */ public void setStepOperations(RepeatOperations stepOperations) { this.stepOperations = stepOperations; @@ -129,7 +127,9 @@ public class SimpleStepExecutor implements StepExecutor { * 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. + * + * @param chunkOperations + * a {@link RepeatOperations} instance. */ public void setChunkOperations(RepeatOperations chunkOperations) { this.chunkOperations = chunkOperations; @@ -144,13 +144,16 @@ public class SimpleStepExecutor implements StepExecutor { * 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 StepInterruptedException + * if the step or a chunk is interrupted + * @throws RuntimeException + * if there is an exception during a chunk execution * @see StepExecutor#process(StepConfiguration, StepExecution) */ - public ExitStatus process(final StepConfiguration configuration, final StepExecution stepExecution) - throws BatchCriticalException, StepInterruptedException { + public ExitStatus process(final StepConfiguration configuration, + final StepExecution stepExecution) throws BatchCriticalException, + StepInterruptedException { final StepInstance step = stepExecution.getStep(); boolean isRestart = step.getStepExecutionCount() > 0 ? true : false; @@ -160,11 +163,17 @@ public class SimpleStepExecutor implements StepExecutor { ExitStatus status = ExitStatus.FAILED; - final SimpleStepContext stepScopeContext = StepSynchronizationManager.open(); + final SimpleStepContext stepScopeContext = StepSynchronizationManager + .open(); stepScopeContext.setStepExecution(stepExecution); + // Add the job identifier so that it can be used to identify + // the conversation in StepScope + stepScopeContext.setAttribute(StepScope.ID_KEY, stepExecution + .getJobExecution().getJobIdentifier()); try { - stepExecution.setStartTime(new Timestamp(System.currentTimeMillis())); + stepExecution + .setStartTime(new Timestamp(System.currentTimeMillis())); updateStatus(stepExecution, BatchStatus.STARTED); final boolean saveRestartData = configuration.isSaveRestartData(); @@ -175,19 +184,18 @@ public class SimpleStepExecutor implements StepExecutor { status = stepOperations.iterate(new RepeatCallback() { - public ExitStatus doInIteration(final RepeatContext context) throws Exception { + public ExitStatus doInIteration(final RepeatContext context) + throws Exception { - stepExecution.getJobExecution().registerStepContext(context); - context.registerDestructionCallback("STEP_EXECUTION_CONTEXT_CALLBACK", new Runnable() { - public void run() { - stepExecution.getJobExecution().unregisterStepContext(context); - } - }); - context.setAttribute(StepScope.ID_KEY, stepExecution.getJobExecution() - .getJobIdentifier()); - // Mark the context as a step context as a hint to scope - // implementations. - context.setAttribute(STEP_KEY, stepExecution); + stepExecution.getJobExecution() + .registerStepContext(context); + context.registerDestructionCallback( + "STEP_EXECUTION_CONTEXT_CALLBACK", new Runnable() { + public void run() { + stepExecution.getJobExecution() + .unregisterStepContext(context); + } + }); // Add the step execution as an attribute so monitoring // clients can see it. context.setAttribute(STEP_EXECUTION_KEY, stepExecution); @@ -195,18 +203,21 @@ public class SimpleStepExecutor implements StepExecutor { // interruption. interruptionPolicy.checkInterrupted(context); - ExitStatus result = (ExitStatus) new TransactionTemplate(transactionManager) + ExitStatus result = (ExitStatus) new TransactionTemplate( + transactionManager) .execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus status) { + public Object doInTransaction( + TransactionStatus status) { // New transaction obtained, resynchronize // TransactionSyncrhonization objects - BatchTransactionSynchronizationManager.resynchronize(); + BatchTransactionSynchronizationManager + .resynchronize(); ExitStatus result; try { - result = processChunk(configuration, stepExecution); - } - catch (Throwable t) { + result = processChunk(configuration, + stepExecution); + } catch (Throwable t) { /* * any exception thrown within the * transaction template will @@ -216,19 +227,20 @@ public class SimpleStepExecutor implements StepExecutor { stepExecution.incrementRollbackCount(); if (t instanceof RuntimeException) { throw (RuntimeException) t; - } - else { + } else { throw new RuntimeException(t); } } if (saveRestartData) { - step.setRestartData(getRestartData(module)); + step + .setRestartData(getRestartData(module)); jobRepository.update(step); } Properties statistics = getStatistics(module); stepExecution.setStatistics(statistics); - context.setAttribute(STATISTICS_KEY, statistics); + context.setAttribute(STATISTICS_KEY, + statistics); stepExecution.incrementCommitCount(); jobRepository.saveOrUpdate(stepExecution); return result; @@ -248,34 +260,30 @@ public class SimpleStepExecutor implements StepExecutor { updateStatus(stepExecution, BatchStatus.COMPLETED); return status; - } - catch (RuntimeException e) { + } catch (RuntimeException e) { - //classify exception so an exit code can be stored. + // 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 { + } else { updateStatus(stepExecution, BatchStatus.FAILED); throw e; } - } - finally { + } finally { stepExecution.setExitStatus(status); stepExecution.setEndTime(new Timestamp(System.currentTimeMillis())); try { jobRepository.saveOrUpdate(stepExecution); - } - finally { + } finally { // clear any registered synchronizations try { StepSynchronizationManager.close(); - } - finally { - BatchTransactionSynchronizationManager.clearSynchronizations(); + } finally { + BatchTransactionSynchronizationManager + .clearSynchronizations(); } } } @@ -284,9 +292,13 @@ public class SimpleStepExecutor implements StepExecutor { /** * Convenience method to update the status in all relevant places. - * @param step the current step - * @param stepExecution the current stepExecution - * @param status the status to set + * + * @param step + * the current step + * @param stepExecution + * the current stepExecution + * @param status + * the status to set */ private void updateStatus(StepExecution stepExecution, BatchStatus status) { StepInstance step = stepExecution.getStep(); @@ -294,7 +306,8 @@ public class SimpleStepExecutor implements StepExecutor { step.setStatus(status); jobRepository.update(step); jobRepository.saveOrUpdate(stepExecution); - for (Iterator iter = stepExecution.getJobExecution().getStepContexts().iterator(); iter.hasNext();) { + for (Iterator iter = stepExecution.getJobExecution().getStepContexts() + .iterator(); iter.hasNext();) { RepeatContext context = (RepeatContext) iter.next(); context.setAttribute("JOB_STATUS", status); } @@ -305,25 +318,32 @@ public class SimpleStepExecutor implements StepExecutor { * transaction. The transaction is programmatically started and stopped * outside this method, so subclasses that override do not need to create a * transaction. - * - * @param configuration the current step configuration - * @param stepExecution the current step, containing the - * {@link Tasklet} with the business logic. + * + * @param configuration + * the current step configuration + * @param stepExecution + * 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 StepConfiguration configuration, + protected final ExitStatus processChunk( + final StepConfiguration configuration, final StepExecution stepExecution) { return chunkOperations.iterate(new RepeatCallback() { - public ExitStatus doInIteration(final RepeatContext context) throws Exception { + public ExitStatus doInIteration(final RepeatContext context) + throws Exception { stepExecution.getJobExecution().registerChunkContext(context); - context.registerDestructionCallback("CHUNK_EXECUTION_CONTEXT_CALLBACK", new Runnable() { - public void run() { - stepExecution.getJobExecution().unregisterStepContext(context); - } - }); + context.registerDestructionCallback( + "CHUNK_EXECUTION_CONTEXT_CALLBACK", new Runnable() { + public void run() { + stepExecution.getJobExecution() + .unregisterStepContext(context); + } + }); // check for interruption before each item as well interruptionPolicy.checkInterrupted(context); - ExitStatus exitStatus = doTaskletProcessing(configuration.getTasklet(), stepExecution.getStep()); + ExitStatus exitStatus = doTaskletProcessing(configuration + .getTasklet(), stepExecution.getStep()); stepExecution.incrementTaskCount(); // check for interruption after each item as well interruptionPolicy.checkInterrupted(context); @@ -336,20 +356,24 @@ public class SimpleStepExecutor implements StepExecutor { * 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. - * @param tasklet the unit of business logic to execute - * @param step the current step + * + * @param tasklet + * the unit of business logic to execute + * @param step + * the current step * @return boolean if there is more processing to do - * @throws Exception if there is an error + * @throws Exception + * if there is an error */ - protected ExitStatus doTaskletProcessing(Tasklet tasklet, StepInstance step) throws Exception { + protected ExitStatus doTaskletProcessing(Tasklet tasklet, StepInstance step) + throws Exception { return tasklet.execute(); } private RestartData getRestartData(Tasklet module) { if (module instanceof Restartable) { return ((Restartable) module).getRestartData(); - } - else { + } else { return null; } } @@ -363,8 +387,7 @@ public class SimpleStepExecutor implements StepExecutor { private Properties getStatistics(Tasklet tasklet) { if (tasklet instanceof StatisticsProvider) { return ((StatisticsProvider) tasklet).getStatistics(); - } - else { + } else { return null; } } @@ -373,16 +396,18 @@ public class SimpleStepExecutor implements StepExecutor { * 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} + * + * @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. - * + * Setter for the {@link ExitCodeExceptionClassifier} that will be used to + * classify any exception that causes a job to fail. + * * @param exceptionClassifier */ public void setExceptionClassifier( diff --git a/execution/src/test/java/org/springframework/batch/execution/step/simple/DefaultStepExecutorTests.java b/execution/src/test/java/org/springframework/batch/execution/step/simple/DefaultStepExecutorTests.java index 5df6ea2de..e433d473a 100644 --- a/execution/src/test/java/org/springframework/batch/execution/step/simple/DefaultStepExecutorTests.java +++ b/execution/src/test/java/org/springframework/batch/execution/step/simple/DefaultStepExecutorTests.java @@ -32,6 +32,7 @@ import org.springframework.batch.core.tasklet.Tasklet; 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.scope.StepScope; import org.springframework.batch.execution.scope.StepSynchronizationManager; import org.springframework.batch.execution.step.SimpleStepConfiguration; import org.springframework.batch.execution.tasklet.ItemProviderProcessTasklet; @@ -183,6 +184,9 @@ public class DefaultStepExecutorTests extends TestCase { .getStepExecution()); assertEquals(stepExecution, StepSynchronizationManager.getContext() .getStepExecution()); + // StepScope can obtain id information.... + assertNotNull(StepSynchronizationManager.getContext() + .getAttribute(StepScope.ID_KEY)); } });