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

IN PROGRESS - BATCH-929: Deferrable Constraints cause unrecoverable errors

Browse Source 
allow commit failure to cause execution context rollback instead of step failure with UNKNOWN status
This commit is contained in:
robokaso
2008-11-19 12:26:17 +00:00
parent edff656577
commit b094db633b
5 changed files with 147 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
spring-batch-core/src/main/java/org/springframework/batch/core/repository/JobRepository.java
View File

@@ -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
*/

8
spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java
View File

@@ -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);
}
}

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

@@ -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<Exception, Boolean> nonFatalCommitExceptions = new Classifier<Exception, Boolean>() {
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<Exception, Boolean> 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);
}

9
spring-batch-core/src/test/java/org/springframework/batch/core/step/JobRepositorySupport.java
View File

@@ -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;
}
}

74
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java
View File

@@ -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<Exception, Boolean>() {
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();
}
}
}