RESOLVED - issue BATCH-546: pull duplicates from TaskletStep and ItemOrientedStep into AbstractStep
This commit is contained in:
robokaso
@@ -15,13 +15,24 @@
|
||||
*/
|
||||
package org.springframework.batch.core.step;
|
||||
|
||||
import java.io.PrintWriter;
|
||||
import java.io.StringWriter;
|
||||
import java.util.Date;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.batch.core.BatchStatus;
|
||||
import org.springframework.batch.core.JobInterruptedException;
|
||||
import org.springframework.batch.core.Step;
|
||||
import org.springframework.batch.core.StepExecution;
|
||||
import org.springframework.batch.core.StepExecutionListener;
|
||||
import org.springframework.batch.core.UnexpectedJobExecutionException;
|
||||
import org.springframework.batch.core.launch.support.ExitCodeMapper;
|
||||
import org.springframework.batch.core.listener.CompositeStepExecutionListener;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.batch.core.repository.NoSuchJobException;
|
||||
import org.springframework.batch.item.ExecutionContext;
|
||||
import org.springframework.batch.repeat.ExitStatus;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
@@ -35,6 +46,13 @@ import org.springframework.util.Assert;
|
||||
*/
|
||||
public abstract class AbstractStep implements Step, InitializingBean {
|
||||
|
||||
/**
|
||||
* Exit code for interrupted status.
|
||||
*/
|
||||
public static final String JOB_INTERRUPTED = "JOB_INTERRUPTED";
|
||||
|
||||
private static final Log logger = LogFactory.getLog(AbstractStep.class);
|
||||
|
||||
protected String name;
|
||||
|
||||
protected int startLimit = Integer.MAX_VALUE;
|
||||
@@ -42,7 +60,7 @@ public abstract class AbstractStep implements Step, InitializingBean {
|
||||
protected boolean allowStartIfComplete;
|
||||
|
||||
private CompositeStepExecutionListener listener = new CompositeStepExecutionListener();
|
||||
|
||||
|
||||
private JobRepository jobRepository;
|
||||
|
||||
/**
|
||||
@@ -51,13 +69,11 @@ public abstract class AbstractStep implements Step, InitializingBean {
|
||||
public AbstractStep() {
|
||||
super();
|
||||
}
|
||||
|
||||
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
Assert.notNull(jobRepository, "JobRepository is mandatory");
|
||||
}
|
||||
|
||||
|
||||
|
||||
public String getName() {
|
||||
return this.name;
|
||||
}
|
||||
@@ -107,8 +123,118 @@ public abstract class AbstractStep implements Step, InitializingBean {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
public abstract void execute(StepExecution stepExecution) throws JobInterruptedException,
|
||||
UnexpectedJobExecutionException;
|
||||
protected abstract ExitStatus doExecute(StepExecution stepExecution) throws Exception;
|
||||
|
||||
protected abstract void open(ExecutionContext ctx) throws Exception;
|
||||
|
||||
protected abstract void close(ExecutionContext ctx) throws Exception;
|
||||
|
||||
/**
|
||||
* Template method for step execution logic - calls abstract methods for
|
||||
* resource initialization ({@link #open(ExecutionContext)}), execution
|
||||
* logic ({@link #doExecute(StepExecution)}) and resource closing ({@link #close(ExecutionContext)}).
|
||||
*/
|
||||
public void execute(StepExecution stepExecution) throws JobInterruptedException, UnexpectedJobExecutionException {
|
||||
stepExecution.setStartTime(new Date());
|
||||
stepExecution.setStatus(BatchStatus.STARTED);
|
||||
|
||||
ExitStatus exitStatus = ExitStatus.FAILED;
|
||||
Exception commitException = null;
|
||||
|
||||
try {
|
||||
getCompositeListener().beforeStep(stepExecution);
|
||||
try {
|
||||
open(stepExecution.getExecutionContext());
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new UnexpectedJobExecutionException("Failed to initialize the step", e);
|
||||
}
|
||||
exitStatus = doExecute(stepExecution);
|
||||
exitStatus = exitStatus.and(getCompositeListener().afterStep(stepExecution));
|
||||
|
||||
try {
|
||||
getJobRepository().saveOrUpdateExecutionContext(stepExecution);
|
||||
stepExecution.setStatus(BatchStatus.COMPLETED);
|
||||
}
|
||||
catch (Exception e) {
|
||||
commitException = e;
|
||||
stepExecution.setStatus(BatchStatus.UNKNOWN);
|
||||
}
|
||||
|
||||
}
|
||||
catch (Throwable e) {
|
||||
|
||||
logger.error("Encountered an error executing the step");
|
||||
stepExecution.setStatus(determineBatchStatus(e));
|
||||
exitStatus = getDefaultExitStatusForFailure(e);
|
||||
|
||||
try {
|
||||
exitStatus = exitStatus.and(getCompositeListener().onErrorInStep(stepExecution, e));
|
||||
}
|
||||
catch (Exception ex) {
|
||||
logger.error("Encountered an error on listener close.", ex);
|
||||
}
|
||||
rethrow(e);
|
||||
}
|
||||
finally {
|
||||
|
||||
stepExecution.setExitStatus(exitStatus);
|
||||
stepExecution.setEndTime(new Date());
|
||||
|
||||
try {
|
||||
getJobRepository().saveOrUpdate(stepExecution);
|
||||
}
|
||||
catch (Exception e) {
|
||||
commitException = e;
|
||||
}
|
||||
|
||||
try {
|
||||
close(stepExecution.getExecutionContext());
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("Exception while closing step's resources", e);
|
||||
throw new UnexpectedJobExecutionException("Exception while closing step's resources", e);
|
||||
}
|
||||
|
||||
if (commitException != null) {
|
||||
logger.error("Encountered an error saving batch meta data."
|
||||
+ "This job is now in an unknown state and should not be restarted.", commitException);
|
||||
throw new UnexpectedJobExecutionException("Encountered an error saving batch meta data.",
|
||||
commitException);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static void rethrow(Throwable e) throws JobInterruptedException {
|
||||
if (e instanceof Error) {
|
||||
throw (Error) e;
|
||||
}
|
||||
if (e instanceof JobInterruptedException) {
|
||||
throw (JobInterruptedException) e;
|
||||
}
|
||||
else if (e.getCause() instanceof JobInterruptedException) {
|
||||
throw (JobInterruptedException) e.getCause();
|
||||
}
|
||||
else if (e instanceof RuntimeException) {
|
||||
throw (RuntimeException) e;
|
||||
}
|
||||
throw new UnexpectedJobExecutionException(e);
|
||||
}
|
||||
|
||||
/**
|
||||
* Determine the step status based on the exception.
|
||||
*/
|
||||
private static BatchStatus determineBatchStatus(Throwable e) {
|
||||
if (e instanceof FatalException) {
|
||||
return BatchStatus.UNKNOWN;
|
||||
}
|
||||
else if (e instanceof JobInterruptedException || e.getCause() instanceof JobInterruptedException) {
|
||||
return BatchStatus.STOPPED;
|
||||
}
|
||||
else {
|
||||
return BatchStatus.FAILED;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Register a step listener for callbacks at the appropriate stages in a
|
||||
@@ -137,7 +263,7 @@ public abstract class AbstractStep implements Step, InitializingBean {
|
||||
protected StepExecutionListener getCompositeListener() {
|
||||
return listener;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Public setter for {@link JobRepository}.
|
||||
*
|
||||
@@ -150,7 +276,44 @@ public abstract class AbstractStep implements Step, InitializingBean {
|
||||
protected JobRepository getJobRepository() {
|
||||
return jobRepository;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/**
|
||||
* Default mapping from throwable to {@link ExitStatus}. Clients can modify
|
||||
* the exit code using a {@link StepExecutionListener}.
|
||||
*
|
||||
* @param ex the cause of the failure
|
||||
* @return an {@link ExitStatus}
|
||||
*/
|
||||
private ExitStatus getDefaultExitStatusForFailure(Throwable ex) {
|
||||
ExitStatus exitStatus;
|
||||
if (ex instanceof JobInterruptedException || ex.getCause() instanceof JobInterruptedException) {
|
||||
exitStatus = new ExitStatus(false, JOB_INTERRUPTED, JobInterruptedException.class.getName());
|
||||
}
|
||||
else if (ex instanceof NoSuchJobException || ex.getCause() instanceof NoSuchJobException) {
|
||||
exitStatus = new ExitStatus(false, ExitCodeMapper.NO_SUCH_JOB);
|
||||
}
|
||||
else {
|
||||
String message = "";
|
||||
if (ex != null) {
|
||||
StringWriter writer = new StringWriter();
|
||||
ex.printStackTrace(new PrintWriter(writer));
|
||||
message = writer.toString();
|
||||
}
|
||||
exitStatus = ExitStatus.FAILED.addExitDescription(message);
|
||||
}
|
||||
|
||||
return exitStatus;
|
||||
}
|
||||
|
||||
/**
|
||||
* Signals a fatal exception - e.g. unable to persist batch metadata or
|
||||
* rollback transaction. Throwing this exception will result in storing
|
||||
* {@link BatchStatus#UNKNOWN} as step's status.
|
||||
*/
|
||||
protected class FatalException extends RuntimeException {
|
||||
public FatalException(String string, Exception e) {
|
||||
super(string, e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -15,26 +15,18 @@
|
||||
*/
|
||||
package org.springframework.batch.core.step.item;
|
||||
|
||||
import java.io.PrintWriter;
|
||||
import java.io.StringWriter;
|
||||
import java.util.Date;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.batch.core.BatchStatus;
|
||||
import org.springframework.batch.core.JobInterruptedException;
|
||||
import org.springframework.batch.core.StepContribution;
|
||||
import org.springframework.batch.core.StepExecution;
|
||||
import org.springframework.batch.core.StepExecutionListener;
|
||||
import org.springframework.batch.core.UnexpectedJobExecutionException;
|
||||
import org.springframework.batch.core.launch.support.ExitCodeMapper;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.batch.core.repository.NoSuchJobException;
|
||||
import org.springframework.batch.core.step.AbstractStep;
|
||||
import org.springframework.batch.core.step.StepExecutionSynchronizer;
|
||||
import org.springframework.batch.core.step.StepExecutionSynchronizerFactory;
|
||||
import org.springframework.batch.core.step.StepInterruptionPolicy;
|
||||
import org.springframework.batch.core.step.ThreadStepInterruptionPolicy;
|
||||
import org.springframework.batch.item.ExecutionContext;
|
||||
import org.springframework.batch.item.ItemReader;
|
||||
import org.springframework.batch.item.ItemStream;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
@@ -73,11 +65,6 @@ public class ItemOrientedStep extends AbstractStep {
|
||||
|
||||
private static final Log logger = LogFactory.getLog(ItemOrientedStep.class);
|
||||
|
||||
/**
|
||||
* Exit code for interrupted status.
|
||||
*/
|
||||
public static final String JOB_INTERRUPTED = "JOB_INTERRUPTED";
|
||||
|
||||
private RepeatOperations chunkOperations = new RepeatTemplate();
|
||||
|
||||
private RepeatOperations stepOperations = new RepeatTemplate();
|
||||
@@ -218,231 +205,98 @@ public class ItemOrientedStep extends AbstractStep {
|
||||
* @throws JobInterruptedException if the step or a chunk is interrupted
|
||||
* @throws RuntimeException if there is an exception during a chunk
|
||||
* execution
|
||||
* @see StepExecutor#execute(StepExecution)
|
||||
*
|
||||
*/
|
||||
public void execute(final StepExecution stepExecution) throws UnexpectedJobExecutionException,
|
||||
JobInterruptedException {
|
||||
public ExitStatus doExecute(final StepExecution stepExecution) throws Exception {
|
||||
stream.update(stepExecution.getExecutionContext());
|
||||
getJobRepository().saveOrUpdateExecutionContext(stepExecution);
|
||||
itemHandler.mark();
|
||||
|
||||
ExitStatus status = ExitStatus.FAILED;
|
||||
final ExceptionHolder fatalException = new ExceptionHolder();
|
||||
|
||||
try {
|
||||
return stepOperations.iterate(new RepeatCallback() {
|
||||
|
||||
stepExecution.setStartTime(new Date(System.currentTimeMillis()));
|
||||
// We need to save the step execution right away, before we start
|
||||
// using its ID. It would be better to make the creation atomic in
|
||||
// the caller.
|
||||
fatalException.setException(updateStatus(stepExecution, BatchStatus.STARTED));
|
||||
public ExitStatus doInIteration(RepeatContext context) throws Exception {
|
||||
final StepContribution contribution = stepExecution.createStepContribution();
|
||||
// Before starting a new transaction, check for
|
||||
// interruption.
|
||||
if (stepExecution.isTerminateOnly()) {
|
||||
context.setTerminateOnly();
|
||||
}
|
||||
interruptionPolicy.checkInterrupted(stepExecution);
|
||||
|
||||
// Execute step level listeners *after* the execution context is
|
||||
// fixed in the step. E.g. ItemStream instances need the the same
|
||||
// reference to the ExecutionContext as the step execution.
|
||||
getCompositeListener().beforeStep(stepExecution);
|
||||
stream.open(stepExecution.getExecutionContext());
|
||||
stream.update(stepExecution.getExecutionContext());
|
||||
getJobRepository().saveOrUpdateExecutionContext(stepExecution);
|
||||
itemHandler.mark();
|
||||
ExitStatus exitStatus = ExitStatus.CONTINUABLE;
|
||||
|
||||
status = stepOperations.iterate(new RepeatCallback() {
|
||||
TransactionStatus transaction = transactionManager.getTransaction(new DefaultTransactionDefinition());
|
||||
|
||||
public ExitStatus doInIteration(final RepeatContext context) throws Exception {
|
||||
try {
|
||||
|
||||
final StepContribution contribution = stepExecution.createStepContribution();
|
||||
// Before starting a new transaction, check for
|
||||
// interruption.
|
||||
if (stepExecution.isTerminateOnly()) {
|
||||
context.setTerminateOnly();
|
||||
}
|
||||
interruptionPolicy.checkInterrupted(stepExecution);
|
||||
|
||||
ExitStatus result = ExitStatus.CONTINUABLE;
|
||||
|
||||
TransactionStatus transaction = transactionManager
|
||||
.getTransaction(new DefaultTransactionDefinition());
|
||||
exitStatus = processChunk(stepExecution, contribution);
|
||||
contribution.incrementCommitCount();
|
||||
|
||||
// If the step operations are asynchronous then we need
|
||||
// to synchronize changes to the step execution (at a
|
||||
// minimum).
|
||||
try {
|
||||
|
||||
result = processChunk(stepExecution, contribution);
|
||||
contribution.incrementCommitCount();
|
||||
|
||||
// If the step operations are asynchronous then we need
|
||||
// to synchronize changes to the step execution (at a
|
||||
// minimum).
|
||||
try {
|
||||
synchronizer.lock(stepExecution);
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
stepExecution.setStatus(BatchStatus.STOPPED);
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
|
||||
// Apply the contribution to the step
|
||||
// only if chunk was successful
|
||||
stepExecution.apply(contribution);
|
||||
|
||||
// Attempt to flush before the step execution and stream
|
||||
// state are updated
|
||||
itemHandler.flush();
|
||||
|
||||
stream.update(stepExecution.getExecutionContext());
|
||||
try {
|
||||
getJobRepository().saveOrUpdateExecutionContext(stepExecution);
|
||||
}
|
||||
catch (Exception e) {
|
||||
fatalException.setException(e);
|
||||
stepExecution.setStatus(BatchStatus.UNKNOWN);
|
||||
throw new CommitFailedException(
|
||||
"Fatal error detected during save of step execution context", e);
|
||||
}
|
||||
|
||||
try {
|
||||
itemHandler.mark();
|
||||
transactionManager.commit(transaction);
|
||||
}
|
||||
catch (Exception e) {
|
||||
fatalException.setException(e);
|
||||
stepExecution.setStatus(BatchStatus.UNKNOWN);
|
||||
throw new CommitFailedException("Fatal error detected during commit", e);
|
||||
}
|
||||
|
||||
synchronizer.lock(stepExecution);
|
||||
}
|
||||
catch (Error e) {
|
||||
processRollback(stepExecution, contribution, fatalException, transaction);
|
||||
throw e;
|
||||
catch (InterruptedException e) {
|
||||
stepExecution.setStatus(BatchStatus.STOPPED);
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
|
||||
// Apply the contribution to the step
|
||||
// only if chunk was successful
|
||||
stepExecution.apply(contribution);
|
||||
|
||||
// Attempt to flush before the step execution and stream
|
||||
// state are updated
|
||||
itemHandler.flush();
|
||||
|
||||
stream.update(stepExecution.getExecutionContext());
|
||||
try {
|
||||
getJobRepository().saveOrUpdateExecutionContext(stepExecution);
|
||||
}
|
||||
catch (Exception e) {
|
||||
processRollback(stepExecution, contribution, fatalException, transaction);
|
||||
throw e;
|
||||
}
|
||||
finally {
|
||||
synchronizer.release(stepExecution);
|
||||
fatalException.setException(e);
|
||||
stepExecution.setStatus(BatchStatus.UNKNOWN);
|
||||
throw new FatalException("Fatal error detected during save of step execution context", e);
|
||||
}
|
||||
|
||||
// Check for interruption after transaction as well, so that
|
||||
// the interrupted exception is correctly propagated up to
|
||||
// caller
|
||||
interruptionPolicy.checkInterrupted(stepExecution);
|
||||
|
||||
return result;
|
||||
try {
|
||||
itemHandler.mark();
|
||||
transactionManager.commit(transaction);
|
||||
}
|
||||
catch (Exception e) {
|
||||
fatalException.setException(e);
|
||||
stepExecution.setStatus(BatchStatus.UNKNOWN);
|
||||
logger.error("Fatal error detected during commit.");
|
||||
throw new FatalException("Fatal error detected during commit", e);
|
||||
}
|
||||
|
||||
}
|
||||
});
|
||||
|
||||
status = status.and(getCompositeListener().afterStep(stepExecution));
|
||||
|
||||
fatalException.setException(updateStatus(stepExecution, BatchStatus.COMPLETED));
|
||||
}
|
||||
catch (CommitFailedException e) {
|
||||
logger.error("Fatal error detected during commit.");
|
||||
throw e;
|
||||
}
|
||||
catch (RuntimeException e) {
|
||||
status = processFailure(stepExecution, fatalException, e);
|
||||
if (e.getCause() instanceof JobInterruptedException) {
|
||||
updateStatus(stepExecution, BatchStatus.STOPPED);
|
||||
throw (JobInterruptedException) e.getCause();
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
catch (Error e) {
|
||||
status = processFailure(stepExecution, fatalException, e);
|
||||
throw e;
|
||||
}
|
||||
finally {
|
||||
|
||||
stepExecution.setExitStatus(status);
|
||||
stepExecution.setEndTime(new Date(System.currentTimeMillis()));
|
||||
|
||||
try {
|
||||
getJobRepository().saveOrUpdate(stepExecution);
|
||||
}
|
||||
catch (RuntimeException e) {
|
||||
String msg = "Fatal error detected during final save of meta data";
|
||||
logger.error(msg, e);
|
||||
if (!fatalException.hasException()) {
|
||||
fatalException.setException(e);
|
||||
catch (Error e) {
|
||||
processRollback(stepExecution, contribution, fatalException, transaction);
|
||||
throw e;
|
||||
}
|
||||
throw new UnexpectedJobExecutionException(msg, fatalException.getException());
|
||||
}
|
||||
|
||||
try {
|
||||
stream.close(stepExecution.getExecutionContext());
|
||||
}
|
||||
catch (RuntimeException e) {
|
||||
String msg = "Fatal error detected during close of streams. "
|
||||
+ "The job execution completed (possibly unsuccessfully but with consistent meta-data).";
|
||||
logger.error(msg, e);
|
||||
if (!fatalException.hasException()) {
|
||||
fatalException.setException(e);
|
||||
catch (Exception e) {
|
||||
processRollback(stepExecution, contribution, fatalException, transaction);
|
||||
throw e;
|
||||
}
|
||||
throw new UnexpectedJobExecutionException(msg, fatalException.getException());
|
||||
finally {
|
||||
synchronizer.release(stepExecution);
|
||||
}
|
||||
|
||||
// Check for interruption after transaction as well, so that
|
||||
// the interrupted exception is correctly propagated up to
|
||||
// caller
|
||||
interruptionPolicy.checkInterrupted(stepExecution);
|
||||
|
||||
return exitStatus;
|
||||
}
|
||||
|
||||
if (fatalException.hasException()) {
|
||||
throw new UnexpectedJobExecutionException("Encountered an error saving batch meta data.",
|
||||
fatalException.getException());
|
||||
}
|
||||
});
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* @param stepExecution the current {@link StepExecution}
|
||||
* @param fatalException the {@link ExceptionHolder} containing information
|
||||
* about failures in meta-data
|
||||
* @param e the cause of teh failure
|
||||
* @return an {@link ExitStatus}
|
||||
*/
|
||||
private ExitStatus processFailure(final StepExecution stepExecution, final ExceptionHolder fatalException,
|
||||
Throwable e) {
|
||||
|
||||
// Default classification marks this as a failure and adds the exception
|
||||
// type and message
|
||||
ExitStatus status = getDefaultExitStatusForFailure(e);
|
||||
|
||||
if (!fatalException.hasException()) {
|
||||
try {
|
||||
// classify exception so an exit code can be stored.
|
||||
status = status.and(getCompositeListener().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;
|
||||
}
|
||||
|
||||
/**
|
||||
* Default mapping from throwable to {@link ExitStatus}. Clients can modify
|
||||
* the exit code using a {@link StepExecutionListener}.
|
||||
*
|
||||
* @param throwable the cause of teh failure
|
||||
* @return an {@link ExitStatus}
|
||||
*/
|
||||
private ExitStatus getDefaultExitStatusForFailure(Throwable throwable) {
|
||||
ExitStatus exitStatus;
|
||||
if (throwable instanceof JobInterruptedException) {
|
||||
exitStatus = new ExitStatus(false, JOB_INTERRUPTED, JobInterruptedException.class.getName());
|
||||
}
|
||||
else if (throwable instanceof NoSuchJobException) {
|
||||
exitStatus = new ExitStatus(false, ExitCodeMapper.NO_SUCH_JOB);
|
||||
}
|
||||
else {
|
||||
String message = "";
|
||||
if (throwable != null) {
|
||||
StringWriter writer = new StringWriter();
|
||||
throwable.printStackTrace(new PrintWriter(writer));
|
||||
message = writer.toString();
|
||||
}
|
||||
exitStatus = ExitStatus.FAILED.addExitDescription(message);
|
||||
}
|
||||
|
||||
return exitStatus;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -475,25 +329,6 @@ public class ItemOrientedStep extends AbstractStep {
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* Convenience method to update the status in all relevant places.
|
||||
*
|
||||
* @param stepInstance the current step
|
||||
* @param stepExecution the current stepExecution
|
||||
* @param status the status to set
|
||||
*/
|
||||
private Exception updateStatus(StepExecution stepExecution, BatchStatus status) {
|
||||
stepExecution.setStatus(status);
|
||||
try {
|
||||
getJobRepository().saveOrUpdate(stepExecution);
|
||||
return null;
|
||||
}
|
||||
catch (Exception e) {
|
||||
return e;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* @param stepExecution
|
||||
* @param contribution
|
||||
@@ -522,7 +357,7 @@ public class ItemOrientedStep extends AbstractStep {
|
||||
*/
|
||||
if (!fatalException.hasException()) {
|
||||
fatalException.setException(e);
|
||||
stepExecution.setStatus(BatchStatus.UNKNOWN);
|
||||
throw new FatalException("Failed while processing rollback", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -545,11 +380,11 @@ public class ItemOrientedStep extends AbstractStep {
|
||||
|
||||
}
|
||||
|
||||
private class CommitFailedException extends RuntimeException {
|
||||
|
||||
public CommitFailedException(String string, Exception e) {
|
||||
super(string, e);
|
||||
}
|
||||
protected void close(ExecutionContext ctx) throws Exception {
|
||||
stream.close(ctx);
|
||||
}
|
||||
|
||||
protected void open(ExecutionContext ctx) throws Exception {
|
||||
stream.open(ctx);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -15,21 +15,14 @@
|
||||
*/
|
||||
package org.springframework.batch.core.step.tasklet;
|
||||
|
||||
import java.util.Date;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.batch.core.BatchStatus;
|
||||
import org.springframework.batch.core.UnexpectedJobExecutionException;
|
||||
import org.springframework.batch.core.JobInterruptedException;
|
||||
import org.springframework.batch.core.Step;
|
||||
import org.springframework.batch.core.StepExecution;
|
||||
import org.springframework.batch.core.StepExecutionListener;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.batch.core.step.AbstractStep;
|
||||
import org.springframework.batch.item.ExecutionContext;
|
||||
import org.springframework.batch.repeat.ExitStatus;
|
||||
import org.springframework.beans.factory.BeanNameAware;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
@@ -45,9 +38,7 @@ import org.springframework.util.Assert;
|
||||
* @author Ben Hale
|
||||
* @author Robert Kasanicky
|
||||
*/
|
||||
public class TaskletStep extends AbstractStep implements Step, InitializingBean, BeanNameAware {
|
||||
|
||||
private static final Log logger = LogFactory.getLog(TaskletStep.class);
|
||||
public class TaskletStep extends AbstractStep implements BeanNameAware {
|
||||
|
||||
private Tasklet tasklet;
|
||||
|
||||
@@ -116,62 +107,21 @@ public class TaskletStep extends AbstractStep implements Step, InitializingBean,
|
||||
this.tasklet = tasklet;
|
||||
}
|
||||
|
||||
public void execute(StepExecution stepExecution) throws JobInterruptedException, UnexpectedJobExecutionException {
|
||||
stepExecution.setStartTime(new Date());
|
||||
stepExecution.setStatus(BatchStatus.STARTED);
|
||||
|
||||
ExitStatus exitStatus = ExitStatus.FAILED;
|
||||
Exception fatalException = null;
|
||||
try {
|
||||
|
||||
getCompositeListener().beforeStep(stepExecution);
|
||||
exitStatus = tasklet.execute();
|
||||
exitStatus = exitStatus.and(getCompositeListener().afterStep(stepExecution));
|
||||
|
||||
try {
|
||||
getJobRepository().saveOrUpdateExecutionContext(stepExecution);
|
||||
stepExecution.setStatus(BatchStatus.COMPLETED);
|
||||
}
|
||||
catch (Exception e) {
|
||||
fatalException = e;
|
||||
stepExecution.setStatus(BatchStatus.UNKNOWN);
|
||||
}
|
||||
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("Encountered an error running the tasklet");
|
||||
stepExecution.setStatus(BatchStatus.FAILED);
|
||||
try {
|
||||
exitStatus = exitStatus.and(getCompositeListener().onErrorInStep(stepExecution, e));
|
||||
}
|
||||
catch (Exception ex) {
|
||||
logger.error("Encountered an error on listener close.", ex);
|
||||
}
|
||||
if (e instanceof JobInterruptedException) {
|
||||
throw (JobInterruptedException) e;
|
||||
}
|
||||
else if (e instanceof RuntimeException) {
|
||||
throw (RuntimeException) e;
|
||||
}
|
||||
throw new UnexpectedJobExecutionException(e);
|
||||
}
|
||||
finally {
|
||||
stepExecution.setExitStatus(exitStatus);
|
||||
stepExecution.setEndTime(new Date());
|
||||
try {
|
||||
getJobRepository().saveOrUpdate(stepExecution);
|
||||
}
|
||||
catch (Exception e) {
|
||||
fatalException = e;
|
||||
}
|
||||
if (fatalException != null) {
|
||||
logger.error("Encountered an error saving batch meta data."
|
||||
+ "This job is now in an unknown state and should not be restarted.", fatalException);
|
||||
throw new UnexpectedJobExecutionException("Encountered an error saving batch meta data.",
|
||||
fatalException);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Delegate to tasklet.
|
||||
*/
|
||||
protected ExitStatus doExecute(StepExecution stepExecution) throws Exception {
|
||||
return tasklet.execute();
|
||||
}
|
||||
|
||||
protected void close(ExecutionContext ctx) throws Exception {
|
||||
}
|
||||
|
||||
protected void open(ExecutionContext ctx) throws Exception {
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user