RESOLVED - issue BATCH-419: JobListener and StepListener should pass in JobExecution/StepExecution in each method

http://jira.springframework.org/browse/BATCH-419

added the StepExecution argument to StepListener#onErrorInStep method

OPEN - issue BATCH-329: Make "VERSION" meaningful? 
http://jira.springframework.org/browse/BATCH-329

MapStepExecutionDao now handles version similarly to JdbcStepExecutionDao

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

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

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

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;

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

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

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

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) {
+	public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) {
-		return stepListener.onErrorInStep(e);
+		return stepListener.onErrorInStep(stepExecution, e);
 	}
 
 	/**

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

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 {
 
@@ -108,4 +109,32 @@ public class MapStepExecutionDaoTests extends TestCase {
 		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
+		}
+
+	}
+
 }

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

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

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