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

RESOLVED - issue BATCH-457: Error handling broken in SimpleJob

Browse Source 
Look out for Error types explicitly and re-throw them without wrapping
This commit is contained in:
dsyer
2008-03-14 08:53:27 +00:00
parent 1a316c1aa9
commit 4fd2968e5e
8 changed files with 302 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java
View File

@@ -35,8 +35,9 @@ import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.repeat.ExitStatus;
/**
* Simple implementation of (@link Job} interface providing the ability to run a {@link JobExecution}. Sequentially
* executes a job by iterating through its list of steps.
* Simple implementation of (@link Job} interface providing the ability to run a
* {@link JobExecution}. Sequentially executes a job by iterating through its
* list of steps.
*
* @author Lucas Ward
* @author Dave Syer
@@ -48,8 +49,8 @@ public class SimpleJob extends AbstractJob {
private CompositeJobListener listener = new CompositeJobListener();
/**
* Public setter for injecting {@link JobListener}s. They will all be given the {@link JobListener} callbacks at
* the appropriate point in the job.
* Public setter for injecting {@link JobListener}s. They will all be given
* the {@link JobListener} callbacks at the appropriate point in the job.
*
* @param listeners the listeners to set.
*/
@@ -69,7 +70,8 @@ public class SimpleJob extends AbstractJob {
}
/**
* Run the specified job by looping through the steps and delegating to the {@link Step}.
* Run the specified job by looping through the steps and delegating to the
* {@link Step}.
*
* @see org.springframework.batch.core.Job#execute(org.springframework.batch.core.JobExecution)
*/
@@ -110,7 +112,8 @@ public class SimpleJob extends AbstractJob {
if (isRestart && lastStepExecution != null) {
currentStepExecution.setExecutionContext(lastStepExecution.getExecutionContext());
} else {
}
else {
currentStepExecution.setExecutionContext(new ExecutionContext());
}
@@ -123,22 +126,27 @@ public class SimpleJob extends AbstractJob {
listener.afterJob(execution);
} catch (JobInterruptedException e) {
}
catch (JobInterruptedException e) {
execution.setStatus(BatchStatus.STOPPED);
rethrow(e);
} catch (Throwable t) {
}
catch (Throwable t) {
execution.setStatus(BatchStatus.FAILED);
rethrow(t);
} finally {
}
finally {
ExitStatus status = ExitStatus.FAILED;
if (startedCount == 0) {
if (steps.size() > 0) {
status = ExitStatus.NOOP
.addExitDescription("All steps already completed. No processing was done.");
} else {
.addExitDescription("All steps already completed. No processing was done.");
}
else {
status = ExitStatus.NOOP.addExitDescription("No steps configured for this job.");
}
} else if (currentStepExecution != null) {
}
else if (currentStepExecution != null) {
status = currentStepExecution.getExitStatus();
}
@@ -155,8 +163,8 @@ public class SimpleJob extends AbstractJob {
}
/*
* Given a step and configuration, return true if the step should start, false if it should not, and throw an
* exception if the job should finish.
* Given a step and configuration, return true if the step should start,
* false if it should not, and throw an exception if the job should finish.
*/
private boolean shouldStart(JobInstance jobInstance, Step step) throws JobExecutionException {
@@ -165,14 +173,15 @@ public class SimpleJob extends AbstractJob {
StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, step);
if (lastStepExecution == null) {
stepStatus = BatchStatus.STARTING;
} else {
}
else {
stepStatus = lastStepExecution.getStatus();
}
if (stepStatus == BatchStatus.UNKNOWN) {
throw new JobExecutionException("Cannot restart step from UNKNOWN status. "
+ "The last execution ended with a failure that could not be rolled back, "
+ "so it may be dangerous to proceed. " + "Manual intervention is probably necessary.");
+ "The last execution ended with a failure that could not be rolled back, "
+ "so it may be dangerous to proceed. " + "Manual intervention is probably necessary.");
}
if (stepStatus == BatchStatus.COMPLETED && step.isAllowStartIfComplete() == false) {
@@ -184,10 +193,11 @@ public class SimpleJob extends AbstractJob {
if (jobRepository.getStepExecutionCount(jobInstance, step) < step.getStartLimit()) {
// step start count is less than start max, return true
return true;
} else {
}
else {
// start max has been exceeded, throw an exception.
throw new UnexpectedJobExecutionException("Maximum start limit exceeded for step: " + step.getName() + "StartMax: "
+ step.getStartLimit());
throw new UnexpectedJobExecutionException("Maximum start limit exceeded for step: " + step.getName()
+ "StartMax: " + step.getStartLimit());
}
}
@@ -197,14 +207,19 @@ public class SimpleJob extends AbstractJob {
private static void rethrow(Throwable t) throws RuntimeException {
if (t instanceof RuntimeException) {
throw (RuntimeException) t;
} else {
}
else if (t instanceof Error) {
throw (Error) t;
}
else {
throw new UnexpectedJobExecutionException(t);
}
}
/**
* Public setter for the {@link JobRepository} that is needed to manage the state of the batch meta domain (jobs,
* steps, executions) during the life of a job.
* Public setter for the {@link JobRepository} that is needed to manage the
* state of the batch meta domain (jobs, steps, executions) during the life
* of a job.
*
* @param jobRepository
*/

255
spring-batch-core/src/main/java/org/springframework/batch/core/step/ItemOrientedStep.java
View File

@@ -42,16 +42,20 @@ import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
/**
* 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).<br/>
* 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).<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.
* 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
@@ -118,10 +122,13 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
* Register each of the streams for callbacks at the appropriate time in the step. The {@link ItemReader} and
* {@link ItemWriter} are automatically registered, but it doesn't hurt to also register them here. Injected
* dependencies of the reader and writer are not automatically registered, so if you implement {@link ItemWriter}
* using delegation to another object which itself is a {@link ItemStream}, you need to register the delegate here.
* Register each of the streams for callbacks at the appropriate time in the
* step. The {@link ItemReader} and {@link ItemWriter} are automatically
* registered, but it doesn't hurt to also register them here. Injected
* dependencies of the reader and writer are not automatically registered,
* so if you implement {@link ItemWriter} using delegation to another object
* which itself is a {@link ItemStream}, you need to register the delegate
* here.
*
* @param streams an array of {@link ItemStream} objects.
*/
@@ -132,7 +139,8 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
* Register a single {@link ItemStream} for callbacks to the stream interface.
* Register a single {@link ItemStream} for callbacks to the stream
* interface.
*
* @param stream
*/
@@ -141,9 +149,11 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
* Register each of the objects as listeners. If the {@link ItemReader} or {@link ItemWriter} themselves implements
* this interface they will be registered automatically, but their injected dependencies will not be. This is a good
* way to get access to job parameters and execution context if the tasklet is parameterised.
* Register each of the objects as listeners. If the {@link ItemReader} or
* {@link ItemWriter} themselves implements this interface they will be
* registered automatically, but their injected dependencies will not be.
* This is a good way to get access to job parameters and execution context
* if the tasklet is parameterised.
*
* @param listeners an array of listener objects of known types.
*/
@@ -154,7 +164,8 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
* 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 StepListener}
*/
@@ -163,8 +174,9 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
* 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.
*/
@@ -173,8 +185,9 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
* 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.
*/
@@ -183,8 +196,9 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
* Setter for the {@link StepInterruptionPolicy}. The policy is used to check whether an external request has been
* made to interrupt the job execution.
* 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}
*/
@@ -193,8 +207,8 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
* Setter for the {@link ExitStatusExceptionClassifier} that will be used to classify any exception that causes a
* job to fail.
* Setter for the {@link ExitStatusExceptionClassifier} that will be used to
* classify any exception that causes a job to fail.
*
* @param exceptionClassifier
*/
@@ -203,8 +217,9 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
* Mostly useful for testing, but could be used to remove dependence on backport concurrency utilities. Public
* setter for the {@link StepExecutionSynchronizer}.
* Mostly useful for testing, but could be used to remove dependence on
* backport concurrency utilities. Public setter for the
* {@link StepExecutionSynchronizer}.
*
* @param synchronizer the {@link StepExecutionSynchronizer} to set
*/
@@ -213,18 +228,22 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
* 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}.<br/>
* 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}.<br/>
*
* @throws JobInterruptedException 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 UnexpectedJobExecutionException,
JobInterruptedException {
JobInterruptedException {
ExitStatus status = ExitStatus.FAILED;
final ExceptionHolder fatalException = new ExceptionHolder();
@@ -256,7 +275,7 @@ public class ItemOrientedStep extends AbstractStep {
ExitStatus result = ExitStatus.CONTINUABLE;
TransactionStatus transaction = transactionManager
.getTransaction(new DefaultTransactionDefinition());
.getTransaction(new DefaultTransactionDefinition());
try {
@@ -269,7 +288,8 @@ public class ItemOrientedStep extends AbstractStep {
// minimum).
try {
synchronizer.lock(stepExecution);
} catch (InterruptedException e) {
}
catch (InterruptedException e) {
stepExecution.setStatus(BatchStatus.STOPPED);
Thread.currentThread().interrupt();
}
@@ -285,51 +305,34 @@ public class ItemOrientedStep extends AbstractStep {
stream.update(stepExecution.getExecutionContext());
try {
jobRepository.saveOrUpdateExecutionContext(stepExecution);
} catch (Exception e) {
}
catch (Exception e) {
fatalException.setException(e);
stepExecution.setStatus(BatchStatus.UNKNOWN);
throw new CommitFailedException(
"Fatal error detected during save of step execution context", e);
"Fatal error detected during save of step execution context", e);
}
try {
itemHandler.mark();
transactionManager.commit(transaction);
} catch (Exception e) {
}
catch (Exception e) {
fatalException.setException(e);
stepExecution.setStatus(BatchStatus.UNKNOWN);
throw new CommitFailedException("Fatal error detected during commit", e);
}
} catch (Throwable t) {
/*
* Any exception thrown within the transaction should automatically cause the transaction to
* rollback.
*/
stepExecution.rollback();
try {
itemHandler.reset();
itemHandler.clear();
transactionManager.rollback(transaction);
} catch (Exception e) {
/*
* If we already failed to commit, it doesn't help to do this again - it's better to allow
* the CommitFailedException to propagate
*/
if (!fatalException.hasException()) {
fatalException.setException(e);
stepExecution.setStatus(BatchStatus.UNKNOWN);
}
}
if (t instanceof RuntimeException) {
throw (RuntimeException) t;
} else {
throw new RuntimeException(t);
}
} finally {
}
catch (Error e) {
processRollback(stepExecution, fatalException, transaction);
throw e;
}
catch (Exception e) {
processRollback(stepExecution, fatalException, transaction);
throw e;
}
finally {
synchronizer.release(stepExecution);
}
@@ -344,35 +347,29 @@ public class ItemOrientedStep extends AbstractStep {
});
fatalException.setException(updateStatus(stepExecution, BatchStatus.COMPLETED));
} catch (CommitFailedException e) {
}
catch (CommitFailedException e) {
logger.error("Fatal error detected during commit.");
throw e;
} catch (RuntimeException e) {
// classify exception so an exit code can be stored.
status = exceptionClassifier.classifyForExitCode(e);
}
catch (RuntimeException e) {
status = processFailure(stepExecution, fatalException, e);
if (e.getCause() instanceof JobInterruptedException) {
updateStatus(stepExecution, BatchStatus.STOPPED);
throw (JobInterruptedException) e.getCause();
} else if (!fatalException.hasException()) {
try {
status = status.and(listener.onErrorInStep(stepExecution, e));
} catch (RuntimeException ex) {
logger.error("Unexpected error in listener on error in step.", ex);
}
updateStatus(stepExecution, BatchStatus.FAILED);
throw e;
} else {
logger.error("Fatal error detected during rollback caused by underlying exception: ", e);
throw e;
}
} finally {
throw e;
}
catch (Error e) {
status = processFailure(stepExecution, fatalException, e);
throw e;
}
finally {
try {
status = status.and(listener.afterStep(stepExecution));
} catch (RuntimeException e) {
}
catch (RuntimeException e) {
logger.error("Unexpected error in listener after step.", e);
}
@@ -381,7 +378,8 @@ public class ItemOrientedStep extends AbstractStep {
try {
jobRepository.saveOrUpdate(stepExecution);
} catch (RuntimeException e) {
}
catch (RuntimeException e) {
String msg = "Fatal error detected during final save of meta data";
logger.error(msg, e);
if (!fatalException.hasException()) {
@@ -392,9 +390,10 @@ public class ItemOrientedStep extends AbstractStep {
try {
stream.close(stepExecution.getExecutionContext());
} catch (RuntimeException e) {
}
catch (RuntimeException e) {
String msg = "Fatal error detected during close of streams. "
+ "The job execution completed (possibly unsuccessfully but with consistent meta-data).";
+ "The job execution completed (possibly unsuccessfully but with consistent meta-data).";
logger.error(msg, e);
if (!fatalException.hasException()) {
fatalException.setException(e);
@@ -404,7 +403,7 @@ public class ItemOrientedStep extends AbstractStep {
if (fatalException.hasException()) {
throw new UnexpectedJobExecutionException("Encountered an error saving batch meta data.",
fatalException.getException());
fatalException.getException());
}
}
@@ -412,11 +411,41 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
* 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
* @param stepExecution
* @param fatalException
* @param e
* @return
* @throws JobInterruptedException
*/
private ExitStatus processFailure(final StepExecution stepExecution, final ExceptionHolder fatalException,
Throwable e) throws JobInterruptedException {
ExitStatus status;
// classify exception so an exit code can be stored.
status = exceptionClassifier.classifyForExitCode(e);
if (!fatalException.hasException()) {
try {
status = status.and(listener.onErrorInStep(stepExecution, e));
}
catch (RuntimeException ex) {
logger.error("Unexpected error in listener on error in step.", ex);
}
updateStatus(stepExecution, BatchStatus.FAILED);
}
else {
logger.error("Fatal error detected during rollback caused by underlying exception: ", e);
}
return status;
}
/**
* 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.
* @param step the current step containing the {@link Tasklet} with the
* business logic.
* @return true if there is more data to process.
*/
protected ExitStatus processChunk(final StepContribution contribution) {
@@ -449,12 +478,44 @@ public class ItemOrientedStep extends AbstractStep {
try {
jobRepository.saveOrUpdate(stepExecution);
return null;
} catch (Exception e) {
}
catch (Exception e) {
return e;
}
}
/**
* @param stepExecution
* @param fatalException
* @param transaction
*/
private void processRollback(final StepExecution stepExecution, final ExceptionHolder fatalException,
TransactionStatus transaction) {
/*
* Any exception thrown within the transaction should
* automatically cause the transaction to rollback.
*/
stepExecution.rollback();
try {
itemHandler.reset();
itemHandler.clear();
transactionManager.rollback(transaction);
}
catch (Exception e) {
/*
* If we already failed to commit, it doesn't help
* to do this again - it's better to allow the
* CommitFailedException to propagate
*/
if (!fatalException.hasException()) {
fatalException.setException(e);
stepExecution.setStatus(BatchStatus.UNKNOWN);
}
}
}
private static class ExceptionHolder {
private Exception exception;

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

@@ -166,7 +166,7 @@ public class TaskletStep extends AbstractStep implements Step, InitializingBean,
if (e instanceof JobInterruptedException) {
throw (JobInterruptedException) e;
}
if (e instanceof RuntimeException) {
else if (e instanceof RuntimeException) {
throw (RuntimeException) e;
}
throw new UnexpectedJobExecutionException(e);