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

REOPENED - BATCH-765: StepExecution should be saved on every commit

Browse Source 
wrapped StepExecution update in try/catch and added testcase for failure scenario
This commit is contained in:
robokaso
2008-08-13 11:27:23 +00:00
parent f291ba3385
commit 23e1d0d01a
2 changed files with 74 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -47,11 +47,12 @@ import org.springframework.transaction.interceptor.TransactionAttribute;
* 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
* 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
@@ -209,9 +210,9 @@ 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
* 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 ExecutionContext}.<br/>
*
@@ -246,14 +247,16 @@ public class ItemOrientedStep extends AbstractStep {
try {
exitStatus = processChunk(stepExecution, contribution);
} catch (Error e) {
}
catch (Error e) {
if (transactionAttribute.rollbackOn(e)) {
throw e;
}
} catch (Exception e) {
}
catch (Exception e) {
if (transactionAttribute.rollbackOn(e)) {
throw e;
}
}
}
contribution.incrementCommitCount();
@@ -277,26 +280,30 @@ public class ItemOrientedStep extends AbstractStep {
// state are updated
try {
itemHandler.flush();
} catch (Error e) {
}
catch (Error e) {
if (transactionAttribute.rollbackOn(e)) {
throw e;
}
} catch (Exception e) {
}
catch (Exception e) {
if (transactionAttribute.rollbackOn(e)) {
throw e;
}
}
}
try {
stream.update(stepExecution.getExecutionContext());
} catch (Error e) {
}
catch (Error e) {
if (transactionAttribute.rollbackOn(e)) {
throw e;
}
} catch (Exception e) {
}
catch (Exception e) {
if (transactionAttribute.rollbackOn(e)) {
throw e;
}
}
}
try {
@@ -318,8 +325,15 @@ public class ItemOrientedStep extends AbstractStep {
logger.error("Fatal error detected during commit.");
throw new FatalException("Fatal error detected during commit", e);
}
getJobRepository().saveOrUpdate(stepExecution);
try {
getJobRepository().saveOrUpdate(stepExecution);
}
catch (Exception e) {
fatalException.setException(e);
stepExecution.setStatus(BatchStatus.UNKNOWN);
throw new FatalException("Fatal error detected during update of step execution", e);
}
}
catch (Error e) {
@@ -399,8 +413,8 @@ public class ItemOrientedStep extends AbstractStep {
}
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 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);

39
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepTests.java
View File

@@ -34,6 +34,7 @@ import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.UnexpectedJobExecutionException;
import org.springframework.batch.core.job.JobSupport;
import org.springframework.batch.core.listener.StepExecutionListenerSupport;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.dao.MapJobExecutionDao;
import org.springframework.batch.core.repository.dao.MapJobInstanceDao;
import org.springframework.batch.core.repository.dao.MapStepExecutionDao;
@@ -58,6 +59,7 @@ import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.batch.support.PropertiesConverter;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.transaction.TransactionException;
import org.springframework.transaction.support.DefaultTransactionStatus;
@@ -810,6 +812,43 @@ public class ItemOrientedStepTests extends TestCase {
assertEquals(3, stepExecution.getItemCount().intValue());
assertTrue(3 <= jobRepository.updateCount);
}
/**
* Failure to update StepExecution after chunk commit is fatal.
*/
public void testStepExecutionUpdateFailure() throws Exception {
JobExecution jobExecution = new JobExecution(jobInstance);
StepExecution stepExecution = new StepExecution(itemOrientedStep.getName(), jobExecution);
JobRepository repository = new JobRepositoryFailedUpdateStub();
itemOrientedStep.setJobRepository(repository);
itemOrientedStep.afterPropertiesSet();
try {
itemOrientedStep.execute(stepExecution);
fail();
}
catch (Exception e) {
assertEquals(BatchStatus.UNKNOWN, stepExecution.getStatus());
assertEquals("Fatal error detected during update of step execution", e.getMessage());
assertEquals("stub exception", e.getCause().getMessage());
}
}
private static class JobRepositoryFailedUpdateStub extends JobRepositorySupport {
private int timesCalled = 0;
public void saveOrUpdate(StepExecution stepExecution) {
timesCalled++;
if (timesCalled == 2) {
throw new DataAccessResourceFailureException("stub exception");
}
}
}
private static class JobRepositoryStub extends JobRepositorySupport {