diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/JobRepository.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/JobRepository.java index 76ca5e21f..1a747f091 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/JobRepository.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/JobRepository.java @@ -39,11 +39,13 @@ import org.springframework.batch.item.ExecutionContext; public interface JobRepository { /** - * Check if an instance of this job already exists with the parameters provided. + * Check if an instance of this job already exists with the parameters + * provided. * * @param jobName the name of the job * @param jobParameters the parameters to match - * @return true if a {@link JobInstance} already exists for this job name and job parameters + * @return true if a {@link JobInstance} already exists for this job name + * and job parameters */ boolean isJobInstanceExists(String jobName, JobParameters jobParameters); @@ -68,8 +70,8 @@ public interface JobRepository { * found and was already completed successfully. * */ - JobExecution createJobExecution(String jobName, JobParameters jobParameters) throws JobExecutionAlreadyRunningException, - JobRestartException, JobInstanceAlreadyCompleteException; + JobExecution createJobExecution(String jobName, JobParameters jobParameters) + throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException; /** * Update the {@link JobExecution}. @@ -112,6 +114,22 @@ public interface JobRepository { */ void updateExecutionContext(StepExecution stepExecution); + /** + * Load the {@link ExecutionContext} of the given {@link StepExecution}. + * + * @param stepExecution the {@link StepExecution} containing the + * {@link ExecutionContext}. + */ + ExecutionContext getExecutionContext(StepExecution stepExecution); + + /** + * Load the {@link ExecutionContext} of the given {@link JobExecution}. + * + * @param jobExecution the {@link JobExecution} containing the + * {@link ExecutionContext}. + */ + ExecutionContext getExecutionContext(JobExecution jobExecution); + /** * @param stepName the name of the step execution that might have run. * @return the last execution of step for the given job instance. @@ -125,7 +143,7 @@ public interface JobRepository { int getStepExecutionCount(JobInstance jobInstance, String stepName); /** - * @param jobName the name of the job that might have run + * @param jobName the name of the job that might have run * @param jobParameters parameters identifying the {@link JobInstance} * @return the last execution of job if exists, null otherwise */ diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java index d358b2039..3d063fab5 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java @@ -336,4 +336,12 @@ public class SimpleJobRepository implements JobRepository { } + public ExecutionContext getExecutionContext(StepExecution stepExecution) { + return ecDao.getExecutionContext(stepExecution); + } + + public ExecutionContext getExecutionContext(JobExecution jobExecution) { + return ecDao.getExecutionContext(jobExecution); + } + } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java index a646e9187..1515c27cd 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java @@ -20,6 +20,7 @@ import java.util.concurrent.Semaphore; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInterruptedException; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.StepExecution; @@ -39,6 +40,7 @@ import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.RepeatOperations; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.batch.repeat.support.RepeatTemplate; +import org.springframework.batch.support.Classifier; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.interceptor.DefaultTransactionAttribute; @@ -88,6 +90,14 @@ public class TaskletStep extends AbstractStep { private Semaphore semaphore = new Semaphore(1); + private Classifier nonFatalCommitExceptions = new Classifier() { + + public Boolean classify(Exception classifiable) { + return false; + } + + }; + /** * Default constructor. */ @@ -102,6 +112,14 @@ public class TaskletStep extends AbstractStep { super(name); } + /** + * @param nonFatalCommitExceptions classifies whether commit exception is + * fatal or not. + */ + public void setNonFatalCommitExceptions(Classifier nonFatalCommitExceptions) { + this.nonFatalCommitExceptions = nonFatalCommitExceptions; + } + /** * Public setter for the {@link PlatformTransactionManager}. * @@ -278,10 +296,18 @@ public class TaskletStep extends AbstractStep { 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); + if (nonFatalCommitExceptions.classify(e)) { + stepExecution.setExecutionContext(getJobRepository().getExecutionContext(stepExecution)); + JobExecution jobExecution = stepExecution.getJobExecution(); + jobExecution.setExecutionContext(getJobRepository().getExecutionContext(jobExecution)); + throw new CommitException("non-fatal commit failure", e); + } + else { + fatalException.setException(e); + stepExecution.setStatus(BatchStatus.UNKNOWN); + logger.error("Fatal error detected during commit."); + throw new FatalException("Fatal error detected during commit", e); + } } try { @@ -300,7 +326,10 @@ public class TaskletStep extends AbstractStep { throw e; } catch (Exception e) { - processRollback(stepExecution, fatalException, transaction); + // if commit failed, calling rollback on tx manager would cause exception + if (!(e instanceof CommitException)) { + processRollback(stepExecution, fatalException, transaction); + } throw e; } finally { @@ -376,6 +405,15 @@ public class TaskletStep extends AbstractStep { } + /** + * Signals non-fatal commit failure. + */ + private static class CommitException extends RuntimeException { + public CommitException(String msg, Throwable cause) { + super(msg, cause); + } + } + protected void close(ExecutionContext ctx) throws Exception { stream.close(ctx); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/JobRepositorySupport.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/JobRepositorySupport.java index 202ed7477..79c8d32f6 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/JobRepositorySupport.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/JobRepositorySupport.java @@ -20,6 +20,7 @@ import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.item.ExecutionContext; /** * @author Dave Syer @@ -82,4 +83,12 @@ public class JobRepositorySupport implements JobRepository { return null; } + public ExecutionContext getExecutionContext(StepExecution stepExecution) { + return null; + } + + public ExecutionContext getExecutionContext(JobExecution jobExecution) { + return null; + } + } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java index d8fbc830c..91bdabc62 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java @@ -11,7 +11,6 @@ import org.junit.Test; import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; -import org.springframework.batch.core.JobInterruptedException; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.StepExecution; @@ -28,6 +27,7 @@ import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemStreamException; import org.springframework.batch.item.ItemStreamSupport; import org.springframework.batch.repeat.RepeatStatus; +import org.springframework.batch.support.Classifier; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; import org.springframework.core.AttributeAccessor; import org.springframework.transaction.TransactionException; @@ -41,15 +41,13 @@ import org.springframework.transaction.support.DefaultTransactionStatus; */ public class TaskletStepExceptionTests { - TaskletStep taskletStep; + private TaskletStep taskletStep; - StepExecution stepExecution; + private StepExecution stepExecution; - UpdateCountingJobRepository jobRepository; + private UpdateCountingJobRepository jobRepository; - static RuntimeException taskletException = new RuntimeException(); - - static JobInterruptedException interruptedException = new JobInterruptedException(""); + private static RuntimeException taskletException = new RuntimeException(); @Before public void init() { @@ -134,7 +132,7 @@ public class TaskletStepExceptionTests { } @Test - /** + /* * Exception in afterStep is ignored (only logged). */ public void testAfterStepFAilure() throws Exception { @@ -196,12 +194,61 @@ public class TaskletStepExceptionTests { assertEquals(exception, e.getCause()); } + /** + * If commit exception isn't fatal step shouldn't complete with UNKNOWN + * status and execution context should be rolled back. + */ + @Test + public void testSkippableCommitError() throws Exception { + + class TestItemStream extends ItemStreamSupport { + private boolean called = false; + + @Override + public void update(ExecutionContext executionContext) throws ItemStreamException { + executionContext.put("key", "value"); + called = true; + } + + } + final TestItemStream stream = new TestItemStream(); + taskletStep.registerStream(stream); + + final RuntimeException commitException = new RuntimeException(); + taskletStep.setNonFatalCommitExceptions(new Classifier() { + + public Boolean classify(Exception classifiable) { + return true; + } + }); + taskletStep.setTransactionManager(new ResourcelessTransactionManager() { + @Override + protected void doCommit(DefaultTransactionStatus status) throws TransactionException { + throw commitException; + } + }); + + taskletStep.setTasklet(new Tasklet() { + + public RepeatStatus execute(StepContribution contribution, AttributeAccessor attributes) throws Exception { + return RepeatStatus.FINISHED; + } + + }); + + taskletStep.execute(stepExecution); + assertEquals("step won't refuse to restart", FAILED, stepExecution.getStatus()); + assertTrue("execution context modified", stream.called); + assertTrue("execution context rolled back", stepExecution.getExecutionContext().isEmpty()); + } + @Test public void testUpdateError() throws Exception { final RuntimeException exception = new RuntimeException(); taskletStep.setJobRepository(new UpdateCountingJobRepository() { boolean firstCall = true; + @Override public void update(StepExecution arg0) { if (firstCall) { @@ -211,7 +258,7 @@ public class TaskletStepExceptionTests { throw exception; } }); - + taskletStep.execute(stepExecution); assertEquals(UNKNOWN, stepExecution.getStatus()); assertTrue(stepExecution.getFailureExceptions().contains(taskletException)); @@ -272,9 +319,16 @@ public class TaskletStepExceptionTests { } public JobExecution getLastJobExecution(String jobName, JobParameters jobParameters) { - // TODO Auto-generated method stub return null; } + + public ExecutionContext getExecutionContext(StepExecution stepExecution) { + return new ExecutionContext(); + } + + public ExecutionContext getExecutionContext(JobExecution jobExecution) { + return new ExecutionContext(); + } } }