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

IN PROGRESS - issue BATCH-121: BatchResourceFactoryBean - is it adding any value?

Browse Source 
http://opensource.atlassian.com/projects/spring/browse/BATCH-121
This commit is contained in:
dsyer
2007-10-05 16:29:29 +00:00
parent efc6247e90
commit 3559ab7908
2 changed files with 110 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

187
execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java
View File

@@ -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).<br/>
*
*
* 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(