From 23e1d0d01ac78ea3a5edcc7f8be359e3ace0cbde Mon Sep 17 00:00:00 2001 From: robokaso Date: Wed, 13 Aug 2008 11:27:23 +0000 Subject: [PATCH] REOPENED - BATCH-765: StepExecution should be saved on every commit wrapped StepExecution update in try/catch and added testcase for failure scenario --- .../core/step/item/ItemOrientedStep.java | 56 ++++++++++++------- .../core/step/item/ItemOrientedStepTests.java | 39 +++++++++++++ 2 files changed, 74 insertions(+), 21 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java index e23aad41f..d985b8451 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java @@ -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).
* * 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}.
* @@ -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); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepTests.java index 3f72cc338..7ccf496b8 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepTests.java @@ -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 {