diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepListener.java b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepListener.java index 590a710ff..5a854d4d2 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepListener.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepListener.java @@ -38,9 +38,10 @@ public interface StepListener extends BatchListener { * {@link ExitStatus#and(ExitStatus)}. * * @param e an exception thrown by the step execution + * * @return an exit status to be combined with the normal one, or null */ - ExitStatus onErrorInStep(Throwable e); + ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e); /** * Give a listener a chance to modify the exit status from a step. The value diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/BatchListenerSupport.java b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/BatchListenerSupport.java index 5e0fa796c..3191ede8b 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/BatchListenerSupport.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/BatchListenerSupport.java @@ -48,7 +48,7 @@ public class BatchListenerSupport implements StepListener, ChunkListener, /* (non-Javadoc) * @see org.springframework.batch.core.domain.StepListener#onErrorInStep(java.lang.Throwable) */ - public ExitStatus onErrorInStep(Throwable e) { + public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) { return null; } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/StepListenerSupport.java b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/StepListenerSupport.java index 1206835b0..5458c6344 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/StepListenerSupport.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/StepListenerSupport.java @@ -41,7 +41,7 @@ public class StepListenerSupport implements StepListener { /* (non-Javadoc) * @see org.springframework.batch.core.domain.StepListener#onError(java.lang.Throwable) */ - public ExitStatus onErrorInStep(Throwable e) { + public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) { return null; } } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/listener/CompositeStepListener.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/listener/CompositeStepListener.java index 7828e1122..677b2c2d9 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/listener/CompositeStepListener.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/listener/CompositeStepListener.java @@ -78,11 +78,11 @@ public class CompositeStepListener implements StepListener { /* (non-Javadoc) * @see org.springframework.batch.core.domain.StepListener#onError(java.lang.Throwable) */ - public ExitStatus onErrorInStep(Throwable e) { + public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) { ExitStatus status = null; for (Iterator iterator = listeners.iterator(); iterator.hasNext();) { StepListener listener = (StepListener) iterator.next(); - ExitStatus close = listener.onErrorInStep(e); + ExitStatus close = listener.onErrorInStep(stepExecution, e); status = status!=null ? status.and(close): close; } return status; diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/MapStepExecutionDao.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/MapStepExecutionDao.java index 890bff432..0559d4653 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/MapStepExecutionDao.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/MapStepExecutionDao.java @@ -23,8 +23,12 @@ import org.springframework.batch.core.domain.Step; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; +import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.util.Assert; +/** + * In-memory implementation of {@link StepExecutionDao}. + */ public class MapStepExecutionDao implements StepExecutionDao { private static Map executionsByJobExecutionId; @@ -47,21 +51,37 @@ public class MapStepExecutionDao implements StepExecutionDao { } public void saveStepExecution(StepExecution stepExecution) { - Assert.notNull(stepExecution.getJobExecutionId()); + Assert.state(stepExecution.getId() == null); + Assert.state(stepExecution.getVersion() == null); + Assert.notNull(stepExecution.getJobExecutionId(), "JobExecution must be saved already."); + Map executions = (Map) executionsByJobExecutionId.get(stepExecution.getJobExecutionId()); if (executions == null) { executions = TransactionAwareProxyFactory.createTransactionalMap(); executionsByJobExecutionId.put(stepExecution.getJobExecutionId(), executions); } + stepExecution.incrementVersion(); stepExecution.setId(new Long(currentId++)); executions.put(stepExecution.getStepName(), stepExecution); } public void updateStepExecution(StepExecution stepExecution) { + Assert.notNull(stepExecution.getJobExecutionId()); + Map executions = (Map) executionsByJobExecutionId.get(stepExecution.getJobExecutionId()); Assert.notNull(executions, "step executions for given job execution are expected to be already saved"); - Assert.notNull(executions.get(stepExecution.getStepName()), "step execution is expected to be already saved"); + + StepExecution persistedExecution = (StepExecution) executions.get(stepExecution.getStepName()); + Assert.notNull(persistedExecution, "step execution is expected to be already saved"); + + if (!persistedExecution.getVersion().equals(stepExecution.getVersion())) { + throw new OptimisticLockingFailureException("Attempt to update step execution id=" + stepExecution.getId() + + " with wrong version (" + stepExecution.getVersion() + "), where current version is " + + persistedExecution.getVersion()); + } + + stepExecution.incrementVersion(); executions.put(stepExecution.getStepName(), stepExecution); } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java index dfe0bceb0..7fe52c848 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java @@ -399,7 +399,7 @@ public class ItemOrientedStep extends AbstractStep { } else if (!fatalException.hasException()) { try { - status = status.and(listener.onErrorInStep(e)); + status = status.and(listener.onErrorInStep(stepExecution, e)); } catch (RuntimeException ex) { logger.error("Unexpected error in listener on error in step.", ex); diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/TaskletStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/TaskletStep.java index d280c138c..4a2156bcb 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/TaskletStep.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/TaskletStep.java @@ -158,7 +158,7 @@ public class TaskletStep extends AbstractStep implements Step, InitializingBean, logger.error("Encountered an error running the tasklet"); updateStatus(stepExecution, BatchStatus.FAILED); try { - exitStatus = exitStatus.and(listener.onErrorInStep(e)); + exitStatus = exitStatus.and(listener.onErrorInStep(stepExecution, e)); } catch (Exception ex) { logger.error("Encountered an error on listener close.", ex); diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ListenerMulticaster.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ListenerMulticaster.java index 5154ddbc9..929670d77 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ListenerMulticaster.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ListenerMulticaster.java @@ -104,8 +104,8 @@ public class ListenerMulticaster implements StepListener, ChunkListener, ItemRea * @return * @see org.springframework.batch.execution.listener.CompositeStepListener#onErrorInStep(java.lang.Throwable) */ - public ExitStatus onErrorInStep(Throwable e) { - return stepListener.onErrorInStep(e); + public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) { + return stepListener.onErrorInStep(stepExecution, e); } /** diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/listener/CompositeStepListenerTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/listener/CompositeStepListenerTests.java index 12e79471c..af580fcfa 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/listener/CompositeStepListenerTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/listener/CompositeStepListenerTests.java @@ -91,12 +91,12 @@ public class CompositeStepListenerTests extends TestCase { */ public void testOnError() { listener.register(new StepListenerSupport() { - public ExitStatus onErrorInStep(Throwable e) { + public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) { list.add("foo"); return null; } }); - listener.onErrorInStep(new RuntimeException()); + listener.onErrorInStep(null, new RuntimeException()); assertEquals(1, list.size()); } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/MapStepExecutionDaoTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/MapStepExecutionDaoTests.java index cb8e8d512..a22bb8ea1 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/MapStepExecutionDaoTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/MapStepExecutionDaoTests.java @@ -29,6 +29,7 @@ import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.execution.job.JobSupport; import org.springframework.batch.execution.step.StepSupport; import org.springframework.batch.item.ExecutionContext; +import org.springframework.dao.OptimisticLockingFailureException; public class MapStepExecutionDaoTests extends TestCase { @@ -65,14 +66,14 @@ public class MapStepExecutionDaoTests extends TestCase { assertEquals(stepExecution, retrieved); assertEquals(BatchStatus.STARTED, retrieved.getStatus()); } - + public void testUpdateExecution() { stepExecution.setStatus(BatchStatus.STARTED); dao.saveStepExecution(stepExecution); - + stepExecution.setStatus(BatchStatus.STOPPED); dao.updateStepExecution(stepExecution); - + StepExecution retrieved = dao.getStepExecution(jobExecution, step); assertEquals(stepExecution, retrieved); assertEquals(BatchStatus.STOPPED, retrieved.getStatus()); @@ -90,7 +91,7 @@ public class MapStepExecutionDaoTests extends TestCase { ExecutionContext retrieved = dao.findExecutionContext(stepExecution); assertEquals(ctx, retrieved); } - + public void testUpdateContext() { ExecutionContext ctx = new ExecutionContext(new HashMap() { { @@ -102,10 +103,38 @@ public class MapStepExecutionDaoTests extends TestCase { ctx.putLong("longKey", 7); dao.saveOrUpdateExecutionContext(stepExecution); - + ExecutionContext retrieved = dao.findExecutionContext(stepExecution); assertEquals(ctx, retrieved); assertEquals(7, retrieved.getLong("longKey")); } + public void testConcurrentModificationException() { + jobInstance = new JobInstance(new Long(1), new JobParameters(), new JobSupport("testJob")); + jobExecution = new JobExecution(jobInstance, new Long(1)); + step = new StepSupport("foo"); + + StepExecution exec1 = new StepExecution(step, jobExecution); + dao.saveStepExecution(exec1); + + StepExecution exec2 = new StepExecution(step, jobExecution); + exec2.setId(exec1.getId()); + + exec2.incrementVersion(); + assertEquals(new Integer(0), exec1.getVersion()); + assertEquals(exec1.getVersion(), exec2.getVersion()); + + dao.updateStepExecution(exec1); + assertEquals(new Integer(1), exec1.getVersion()); + + try { + dao.updateStepExecution(exec2); + fail(); + } + catch (OptimisticLockingFailureException e) { + // expected + } + + } + } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java index 9be27fbd4..cbf2ea147 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java @@ -417,7 +417,7 @@ public class ItemOrientedStepTests extends TestCase { public void testDirectlyInjectedListenerOnError() throws Exception { itemOrientedStep.registerStepListener(new StepListenerSupport() { - public ExitStatus onErrorInStep(Throwable e) { + public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) { list.add(e); return null; } @@ -706,7 +706,7 @@ public class ItemOrientedStepTests extends TestCase { public void beforeStep(StepExecution stepExecution) { } - public ExitStatus onErrorInStep(Throwable e) { + public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) { return null; } diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java index d529a7c6c..7433668b6 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java @@ -239,7 +239,7 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite /* (non-Javadoc) * @see org.springframework.batch.core.domain.StepListener#onErrorInStep(java.lang.Throwable) */ - public ExitStatus onErrorInStep(Throwable e) { + public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) { return null; } diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/StagingItemWriter.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/StagingItemWriter.java index 485afee1c..5be3e90eb 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/StagingItemWriter.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/StagingItemWriter.java @@ -110,7 +110,7 @@ public class StagingItemWriter extends JdbcDaoSupport implements StepListener, I /* (non-Javadoc) * @see org.springframework.batch.core.domain.StepListener#onErrorInStep(java.lang.Throwable) */ - public ExitStatus onErrorInStep(Throwable e) { + public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) { return null; }