IN PROGRESS - BATCH-929: Deferrable Constraints cause unrecoverable errors
reverted the previous changes, they wouldn't really solve the problem
This commit is contained in:
@@ -39,13 +39,11 @@ 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);
|
||||
|
||||
@@ -70,8 +68,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}.
|
||||
@@ -114,22 +112,6 @@ 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.
|
||||
@@ -143,7 +125,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
|
||||
*/
|
||||
|
||||
@@ -336,12 +336,4 @@ public class SimpleJobRepository implements JobRepository {
|
||||
|
||||
}
|
||||
|
||||
public ExecutionContext getExecutionContext(StepExecution stepExecution) {
|
||||
return ecDao.getExecutionContext(stepExecution);
|
||||
}
|
||||
|
||||
public ExecutionContext getExecutionContext(JobExecution jobExecution) {
|
||||
return ecDao.getExecutionContext(jobExecution);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -48,8 +48,6 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
|
||||
|
||||
private int skipLimit = 0;
|
||||
|
||||
private Collection<Class<? extends Throwable>> nonFatalCommitExceptionClasses = new HashSet<Class<? extends Throwable>>();
|
||||
|
||||
private Collection<Class<? extends Throwable>> skippableExceptionClasses = new HashSet<Class<? extends Throwable>>();
|
||||
|
||||
private Collection<Class<? extends Throwable>> fatalExceptionClasses = new HashSet<Class<? extends Throwable>>();
|
||||
@@ -79,17 +77,6 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
|
||||
this.isReaderTransactionalQueue = isReaderTransactionalQueue;
|
||||
}
|
||||
|
||||
/**
|
||||
* Exceptions that will not cause UNKNOWN status of the step if they occur
|
||||
* on commit. The assumption is that failed commit with one of the supplied
|
||||
* exceptions is equivalent to rollback (and therefore need not cause
|
||||
* failure if skippable).
|
||||
* @param nonFatalCommitExceptionClasses
|
||||
*/
|
||||
public void setNonFatalCommitExceptionClasses(Collection<Class<? extends Throwable>> nonFatalCommitExceptionClasses) {
|
||||
this.nonFatalCommitExceptionClasses = nonFatalCommitExceptionClasses;
|
||||
}
|
||||
|
||||
/**
|
||||
* Setter for the retry policy. If this is specified the other retry
|
||||
* properties are ignored (retryLimit, backOffPolicy,
|
||||
@@ -292,9 +279,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
|
||||
|
||||
step.setTasklet(tasklet);
|
||||
}
|
||||
ItemSkipPolicy commitSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit,
|
||||
nonFatalCommitExceptionClasses, fatalExceptionClasses);
|
||||
step.setCommitSkipPolicy(commitSkipPolicy);
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -20,7 +20,6 @@ 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;
|
||||
@@ -31,8 +30,6 @@ import org.springframework.batch.core.scope.StepContextRepeatCallback;
|
||||
import org.springframework.batch.core.step.AbstractStep;
|
||||
import org.springframework.batch.core.step.StepInterruptionPolicy;
|
||||
import org.springframework.batch.core.step.ThreadStepInterruptionPolicy;
|
||||
import org.springframework.batch.core.step.skip.ItemSkipPolicy;
|
||||
import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy;
|
||||
import org.springframework.batch.item.ExecutionContext;
|
||||
import org.springframework.batch.item.ItemReader;
|
||||
import org.springframework.batch.item.ItemStream;
|
||||
@@ -91,8 +88,6 @@ public class TaskletStep extends AbstractStep {
|
||||
|
||||
private Semaphore semaphore = new Semaphore(1);
|
||||
|
||||
private ItemSkipPolicy commitSkipPolicy = new NeverSkipItemSkipPolicy();
|
||||
|
||||
/**
|
||||
* Default constructor.
|
||||
*/
|
||||
@@ -107,13 +102,6 @@ public class TaskletStep extends AbstractStep {
|
||||
super(name);
|
||||
}
|
||||
|
||||
/**
|
||||
* Skip policy applying to exception thrown on tx commit.
|
||||
*/
|
||||
public void setCommitSkipPolicy(ItemSkipPolicy commitSkipPolicy) {
|
||||
this.commitSkipPolicy = commitSkipPolicy;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the {@link PlatformTransactionManager}.
|
||||
*
|
||||
@@ -290,16 +278,10 @@ public class TaskletStep extends AbstractStep {
|
||||
transactionManager.commit(transaction);
|
||||
}
|
||||
catch (Exception e) {
|
||||
if (commitSkipPolicy.shouldSkip(e, stepExecution.getSkipCount())) {
|
||||
rollbackExecutionContext(stepExecution);
|
||||
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);
|
||||
}
|
||||
fatalException.setException(e);
|
||||
stepExecution.setStatus(BatchStatus.UNKNOWN);
|
||||
logger.error("Fatal error detected during commit.");
|
||||
throw new FatalException("Fatal error detected during commit", e);
|
||||
}
|
||||
|
||||
try {
|
||||
@@ -318,15 +300,7 @@ public class TaskletStep extends AbstractStep {
|
||||
throw e;
|
||||
}
|
||||
catch (Exception e) {
|
||||
// if commit failed, calling rollback on tx manager would
|
||||
// cause exception
|
||||
if (!(e instanceof CommitException)) {
|
||||
processRollback(stepExecution, fatalException, transaction);
|
||||
}
|
||||
else {
|
||||
// assume the failed commit caused rollback
|
||||
stepExecution.rollback();
|
||||
}
|
||||
processRollback(stepExecution, fatalException, transaction);
|
||||
throw e;
|
||||
}
|
||||
finally {
|
||||
@@ -345,15 +319,6 @@ public class TaskletStep extends AbstractStep {
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* Load the saved value of ExecutionContext from repository.
|
||||
*/
|
||||
private void rollbackExecutionContext(StepExecution stepExecution) {
|
||||
stepExecution.setExecutionContext(getJobRepository().getExecutionContext(stepExecution));
|
||||
JobExecution jobExecution = stepExecution.getJobExecution();
|
||||
jobExecution.setExecutionContext(getJobRepository().getExecutionContext(jobExecution));
|
||||
}
|
||||
|
||||
});
|
||||
|
||||
}
|
||||
@@ -411,15 +376,6 @@ 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);
|
||||
}
|
||||
|
||||
@@ -20,7 +20,6 @@ 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
|
||||
@@ -83,12 +82,4 @@ public class JobRepositorySupport implements JobRepository {
|
||||
return null;
|
||||
}
|
||||
|
||||
public ExecutionContext getExecutionContext(StepExecution stepExecution) {
|
||||
return null;
|
||||
}
|
||||
|
||||
public ExecutionContext getExecutionContext(JobExecution jobExecution) {
|
||||
return null;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -525,10 +525,6 @@ public class FaultTolerantStepFactoryBeanTests {
|
||||
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testNonFatalCommitFailure() throws Exception {
|
||||
//TODO
|
||||
}
|
||||
private static class SkipProcessorStub implements ItemProcessor<String, String> {
|
||||
|
||||
private final Collection<String> failures;
|
||||
|
||||
@@ -3,19 +3,15 @@
|
||||
*/
|
||||
package org.springframework.batch.core.step.item;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.springframework.batch.core.BatchStatus.COMPLETED;
|
||||
import static org.springframework.batch.core.BatchStatus.FAILED;
|
||||
import static org.springframework.batch.core.BatchStatus.STOPPED;
|
||||
import static org.springframework.batch.core.BatchStatus.UNKNOWN;
|
||||
import static org.junit.Assert.*;
|
||||
import static org.springframework.batch.core.BatchStatus.*;
|
||||
|
||||
import org.junit.Before;
|
||||
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;
|
||||
@@ -25,7 +21,6 @@ import org.springframework.batch.core.repository.JobExecutionAlreadyRunningExcep
|
||||
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.batch.core.repository.JobRestartException;
|
||||
import org.springframework.batch.core.step.skip.AlwaysSkipItemSkipPolicy;
|
||||
import org.springframework.batch.core.step.tasklet.Tasklet;
|
||||
import org.springframework.batch.core.step.tasklet.TaskletStep;
|
||||
import org.springframework.batch.item.ExecutionContext;
|
||||
@@ -46,13 +41,15 @@ import org.springframework.transaction.support.DefaultTransactionStatus;
|
||||
*/
|
||||
public class TaskletStepExceptionTests {
|
||||
|
||||
private TaskletStep taskletStep;
|
||||
TaskletStep taskletStep;
|
||||
|
||||
private StepExecution stepExecution;
|
||||
StepExecution stepExecution;
|
||||
|
||||
private UpdateCountingJobRepository jobRepository;
|
||||
UpdateCountingJobRepository jobRepository;
|
||||
|
||||
private static RuntimeException taskletException = new RuntimeException();
|
||||
static RuntimeException taskletException = new RuntimeException();
|
||||
|
||||
static JobInterruptedException interruptedException = new JobInterruptedException("");
|
||||
|
||||
@Before
|
||||
public void init() {
|
||||
@@ -137,7 +134,7 @@ public class TaskletStepExceptionTests {
|
||||
}
|
||||
|
||||
@Test
|
||||
/*
|
||||
/**
|
||||
* Exception in afterStep is ignored (only logged).
|
||||
*/
|
||||
public void testAfterStepFAilure() throws Exception {
|
||||
@@ -199,58 +196,12 @@ 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 testNonFatalCommitError() 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.setCommitSkipPolicy(new AlwaysSkipItemSkipPolicy());
|
||||
|
||||
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());
|
||||
assertEquals("failed commit counted as rollback", 1, stepExecution.getRollbackCount());
|
||||
}
|
||||
|
||||
@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) {
|
||||
@@ -260,7 +211,7 @@ public class TaskletStepExceptionTests {
|
||||
throw exception;
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
taskletStep.execute(stepExecution);
|
||||
assertEquals(UNKNOWN, stepExecution.getStatus());
|
||||
assertTrue(stepExecution.getFailureExceptions().contains(taskletException));
|
||||
@@ -321,16 +272,9 @@ 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();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -23,7 +23,6 @@ import org.springframework.batch.core.repository.JobExecutionAlreadyRunningExcep
|
||||
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.batch.core.repository.JobRestartException;
|
||||
import org.springframework.batch.item.ExecutionContext;
|
||||
|
||||
/**
|
||||
* @author Dave Syer
|
||||
@@ -88,12 +87,4 @@ public class JobRepositorySupport implements JobRepository {
|
||||
return null;
|
||||
}
|
||||
|
||||
public ExecutionContext getExecutionContext(StepExecution stepExecution) {
|
||||
return null;
|
||||
}
|
||||
|
||||
public ExecutionContext getExecutionContext(JobExecution jobExecution) {
|
||||
return null;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user