OPEN - issue BATCH-385: Merge user attributes in StepContext with ExecutionContext

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

Make Job responsible for setting up step context.  Javadocs on StepContext to indicate that the attributes are volatile.  Add JobParametersAware (might be temporary).

spring-batch-execution/src/main/java/org/springframework/batch/execution/job/SimpleJob.java

@@ -29,6 +29,9 @@ import org.springframework.batch.core.domain.Step;
 import org.springframework.batch.core.domain.StepExecution;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.runtime.ExitStatusExceptionClassifier;
+import org.springframework.batch.execution.scope.SimpleStepContext;
+import org.springframework.batch.execution.scope.StepContext;
+import org.springframework.batch.execution.scope.StepSynchronizationManager;
 import org.springframework.batch.execution.step.support.SimpleExitStatusExceptionClassifier;
 import org.springframework.batch.io.exception.BatchCriticalException;
 import org.springframework.batch.repeat.ExitStatus;
@@ -79,11 +82,23 @@ public class SimpleJob extends AbstractJob {
 				Step step = (Step) i.next();
 
 				if (shouldStart(jobInstance, step)) {
+
 					startedCount++;
 					updateStatus(execution, BatchStatus.STARTED);
 					StepExecution stepExecution = execution.createStepExecution(step);
-					step.execute(stepExecution);
+
+					StepContext parentStepContext = StepSynchronizationManager.getContext();
+					final StepContext stepContext = new SimpleStepContext(stepExecution, parentStepContext);
+					StepSynchronizationManager.register(stepContext);
+					try {
+						step.execute(stepExecution);
+					} finally {
+						// clear any registered synchronizations
+						StepSynchronizationManager.close();
+					}
+
 					status = stepExecution.getExitStatus();
+
 				}
 			}
 

spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/JobParametersAware.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2006-2007 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.execution.scope;
+
+import org.springframework.batch.core.domain.JobParameters;
+
+/**
+ * Marker interface for callback injecting {@link JobParameters}. A Spring bean
+ * which is step scoped will be injected with the {@link JobParameters} when it
+ * is instantiated. In most cases this will require the use of
+ * <aop:scoped-proxy> when the bean is used as a dependency in a
+ * singleton.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public interface JobParametersAware {
+
+	/**
+	 * Callback method for injection of {@link JobParameters}.
+	 * 
+	 * @param jobParameters the {@link JobParameters} to set.
+	 */
+	void setJobParameters(JobParameters jobParameters);
+
+}

spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/StepContext.java

@@ -16,10 +16,18 @@
 package org.springframework.batch.execution.scope;
 
 import org.springframework.batch.core.domain.StepExecution;
+import org.springframework.batch.item.ExecutionContext;
 import org.springframework.core.AttributeAccessor;
 
 /**
- * Interface for step-scoped context object and step-scoped services.
+ * Interface for step-scoped context object and step-scoped services. This
+ * interface extends {@link AttributeAccessor}, so there is an underlying map
+ * that can be used for storing state during a step execution. The storage is
+ * <em>volatile</em>: the attributes are not persisted and not durable across
+ * steps in a job, or across restarts of a failed job.
+ * 
+ * @see ExecutionContext for access to durable attributes that will be restored
+ * in the case of a restart.
  * 
  * @author Dave Syer
  * 

spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/StepScope.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.batch.execution.scope;
 
+import org.springframework.batch.core.domain.JobParameters;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.ObjectFactory;
 import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
@@ -44,7 +45,8 @@ public class StepScope implements Scope, BeanFactoryPostProcessor, Ordered {
 		this.order = order;
 	}
 
-	/* (non-Javadoc)
+	/*
+	 * (non-Javadoc)
 	 * @see org.springframework.core.Ordered#getOrder()
 	 */
 	public int getOrder() {
@@ -70,6 +72,16 @@ public class StepScope implements Scope, BeanFactoryPostProcessor, Ordered {
 					if (scopedObject instanceof StepContextAware) {
 						((StepContextAware) scopedObject).setStepContext(context);
 					}
+					if (scopedObject instanceof JobParametersAware) {
+						try {
+							JobParameters jobParameters = context.getStepExecution().getJobExecution().getJobInstance()
+									.getJobParameters();
+							((JobParametersAware) scopedObject).setJobParameters(jobParameters);
+						}
+						catch (NullPointerException e) {
+							// ignore
+						}
+					}
 					context.setAttribute(name, scopedObject);
 				}
 			}

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java

@@ -27,9 +27,6 @@ import org.springframework.batch.core.domain.StepExecution;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.runtime.ExitStatusExceptionClassifier;
 import org.springframework.batch.core.tasklet.Tasklet;
-import org.springframework.batch.execution.scope.SimpleStepContext;
-import org.springframework.batch.execution.scope.StepContext;
-import org.springframework.batch.execution.scope.StepSynchronizationManager;
 import org.springframework.batch.execution.step.support.SimpleExitStatusExceptionClassifier;
 import org.springframework.batch.execution.step.support.StepInterruptionPolicy;
 import org.springframework.batch.execution.step.support.ThreadStepInterruptionPolicy;
@@ -233,9 +230,6 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
 			// the caller.
 			fatalException.setException(updateStatus(stepExecution, BatchStatus.STARTED));
 
-			StepContext parentStepContext = StepSynchronizationManager.getContext();
-			final StepContext stepContext = new SimpleStepContext(stepExecution, parentStepContext);
-			StepSynchronizationManager.register(stepContext);
 			possiblyRegisterStreams();
 
 			if (isRestart && lastStepExecution != null) {
@@ -328,8 +322,8 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
 							fatalException.setException(e);
 							stepExecution.setStatus(BatchStatus.UNKNOWN);
 						}
-						
-						if(itemSkipPolicy.shouldFail(t)){
+
+						if (itemSkipPolicy.shouldFail(t)) {
 							if (t instanceof RuntimeException) {
 								throw (RuntimeException) t;
 							}
@@ -337,10 +331,10 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
 								throw new RuntimeException(t);
 							}
 						}
-						else{
+						else {
 							logger.error("Exception should not cause step to fail", t);
 						}
-						
+
 						result = ExitStatus.CONTINUABLE;
 					}
 
@@ -379,40 +373,33 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
 
 		}
 		finally {
+
 			stepExecution.setExitStatus(status);
 			stepExecution.setEndTime(new Date(System.currentTimeMillis()));
 
 			try {
-
-				try {
-					jobRepository.saveOrUpdate(stepExecution);
-				}
-				catch (RuntimeException e) {
-					String msg = "Fatal error detected during final save of meta data";
-					logger.error(msg, e);
-					if (!fatalException.hasException()) {
-						fatalException.setException(e);
-					}
-					throw new BatchCriticalException(msg, fatalException.getException());
-				}
-
-				try {
-					streamManager.close(stepExecution.getExecutionContext());
-				}
-				catch (RuntimeException e) {
-					String msg = "Fatal error detected during close of streams. "
-							+ "The job execution completed (possibly unsuccessfully but with consistent meta-data).";
-					logger.error(msg, e);
-					if (!fatalException.hasException()) {
-						fatalException.setException(e);
-					}
-					throw new BatchCriticalException(msg, fatalException.getException());
-				}
-
+				jobRepository.saveOrUpdate(stepExecution);
 			}
-			finally {
-				// clear any registered synchronizations
-				StepSynchronizationManager.close();
+			catch (RuntimeException e) {
+				String msg = "Fatal error detected during final save of meta data";
+				logger.error(msg, e);
+				if (!fatalException.hasException()) {
+					fatalException.setException(e);
+				}
+				throw new BatchCriticalException(msg, fatalException.getException());
+			}
+
+			try {
+				streamManager.close(stepExecution.getExecutionContext());
+			}
+			catch (RuntimeException e) {
+				String msg = "Fatal error detected during close of streams. "
+						+ "The job execution completed (possibly unsuccessfully but with consistent meta-data).";
+				logger.error(msg, e);
+				if (!fatalException.hasException()) {
+					fatalException.setException(e);
+				}
+				throw new BatchCriticalException(msg, fatalException.getException());
 			}
 
 			if (fatalException.hasException()) {

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/TaskletStep.java

@@ -25,9 +25,6 @@ import org.springframework.batch.core.domain.Step;
 import org.springframework.batch.core.domain.StepExecution;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.tasklet.Tasklet;
-import org.springframework.batch.execution.scope.SimpleStepContext;
-import org.springframework.batch.execution.scope.StepContext;
-import org.springframework.batch.execution.scope.StepSynchronizationManager;
 import org.springframework.batch.io.exception.BatchCriticalException;
 import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.batch.repeat.RepeatCallback;
@@ -53,7 +50,7 @@ public class TaskletStep implements Step, InitializingBean, BeanNameAware {
 	private Tasklet tasklet;
 
 	private JobRepository jobRepository;
-	
+
 	private String name;
 
 	private int startLimit = Integer.MAX_VALUE;
@@ -65,9 +62,11 @@ public class TaskletStep implements Step, InitializingBean, BeanNameAware {
 	}
 
 	/**
-	 * Set the name property if it is not already set. Because of the order of the callbacks in a Spring container the
-	 * name property will be set first if it is present. Care is needed with bean definition inheritance - if a parent
-	 * bean has a name, then its children need an explicit name as well, otherwise they will not be unique.
+	 * Set the name property if it is not already set. Because of the order of
+	 * the callbacks in a Spring container the name property will be set first
+	 * if it is present. Care is needed with bean definition inheritance - if a
+	 * parent bean has a name, then its children need an explicit name as well,
+	 * otherwise they will not be unique.
 	 * 
 	 * @see org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang.String)
 	 */
@@ -78,7 +77,8 @@ public class TaskletStep implements Step, InitializingBean, BeanNameAware {
 	}
 
 	/**
-	 * Set the name property. Always overrides the default value if this object is a Spring bean.
+	 * Set the name property. Always overrides the default value if this object
+	 * is a Spring bean.
 	 * 
 	 * @see #setBeanName(java.lang.String)
 	 */
@@ -112,7 +112,6 @@ public class TaskletStep implements Step, InitializingBean, BeanNameAware {
 		this.allowStartIfComplete = allowStartIfComplete;
 	}
 
-
 	private RepeatListener[] listeners = new RepeatListener[] {};
 
 	public void setListeners(RepeatListener[] listeners) {
@@ -176,10 +175,6 @@ public class TaskletStep implements Step, InitializingBean, BeanNameAware {
 		Exception fatalException = null;
 		try {
 
-			StepContext parentStepContext = StepSynchronizationManager.getContext();
-			final StepContext stepContext = new SimpleStepContext(stepExecution, parentStepContext);
-			StepSynchronizationManager.register(stepContext);
-
 			// We are using the RepeatTemplate as a vehicle for the listener
 			// so it can be set up cheaply here with standard properties.
 			RepeatTemplate template = new RepeatTemplate();
@@ -195,7 +190,8 @@ public class TaskletStep implements Step, InitializingBean, BeanNameAware {
 			try {
 				jobRepository.saveOrUpdateExecutionContext(stepExecution);
 				updateStatus(stepExecution, BatchStatus.COMPLETED);
-			} catch (Exception e) {
+			}
+			catch (Exception e) {
 				fatalException = e;
 				updateStatus(stepExecution, BatchStatus.UNKNOWN);
 			}
@@ -220,16 +216,13 @@ public class TaskletStep implements Step, InitializingBean, BeanNameAware {
 			catch (Exception e) {
 				fatalException = e;
 			}
-			finally {
-				StepSynchronizationManager.close();
-				if (fatalException!=null) {
-					logger.error("Encountered an error saving batch meta data."
-							+ "This job is now in an unknown state and should not be restarted.", fatalException);
-					throw new BatchCriticalException("Encountered an error saving batch meta data.", fatalException);
-				}
+			if (fatalException != null) {
+				logger.error("Encountered an error saving batch meta data."
+						+ "This job is now in an unknown state and should not be restarted.", fatalException);
+				throw new BatchCriticalException("Encountered an error saving batch meta data.", fatalException);
 			}
-		}		
-	
+		}
+
 	}
 
 	private void updateStatus(StepExecution stepExecution, BatchStatus status) {

spring-batch-execution/src/test/java/org/springframework/batch/execution/job/SimpleJobTests.java

@@ -37,6 +37,7 @@ import org.springframework.batch.execution.repository.dao.MapJobExecutionDao;
 import org.springframework.batch.execution.repository.dao.MapJobInstanceDao;
 import org.springframework.batch.execution.repository.dao.MapStepExecutionDao;
 import org.springframework.batch.execution.repository.dao.StepExecutionDao;
+import org.springframework.batch.execution.scope.StepSynchronizationManager;
 import org.springframework.batch.execution.step.AbstractStep;
 import org.springframework.batch.io.exception.BatchCriticalException;
 import org.springframework.batch.item.reader.AbstractItemReader;
@@ -252,6 +253,21 @@ public class SimpleJobTests extends TestCase {
 				"JobInterruptedException"));
 	}
 
+	public void testStepContextInitialized() throws Exception {
+
+		stepConfiguration1.setCallback(new Runnable() {
+			public void run() {
+				assertNotNull(StepSynchronizationManager.getContext().getStepExecution());	
+				list.add("asserted context");
+			};
+		});
+
+		job.execute(jobExecution);
+		assertEquals(2, list.size());
+		assertTrue(list.contains("asserted context"));
+
+	}
+
 	/*
 	 * Check JobRepository to ensure status is being saved.
 	 */

spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/JobParametersAwareStepScopeTests.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2006-2007 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.execution.scope;
+
+import junit.framework.TestCase;
+
+import org.springframework.batch.core.domain.JobExecution;
+import org.springframework.batch.core.domain.JobInstance;
+import org.springframework.batch.core.domain.JobParameters;
+import org.springframework.batch.execution.job.JobSupport;
+import org.springframework.batch.execution.step.StepSupport;
+import org.springframework.batch.repeat.synch.RepeatSynchronizationManager;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.ObjectFactory;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class JobParametersAwareStepScopeTests extends TestCase {
+	
+	private StepScope scope = new StepScope();
+
+	private SimpleStepContext context;
+
+	JobParameters parameters = new JobParameters();
+
+	/*
+	 * (non-Javadoc)
+	 * @see junit.framework.TestCase#setUp()
+	 */
+	protected void setUp() throws Exception {
+		super.setUp();
+		JobExecution jobExecution = new JobExecution(new JobInstance(new Long(1L), parameters, new JobSupport()), new Long(11L));
+		context = new SimpleStepContext(jobExecution.createStepExecution(new StepSupport()));
+		StepSynchronizationManager.register(context);
+	}
+	
+	/* (non-Javadoc)
+	 * @see junit.framework.TestCase#tearDown()
+	 */
+	protected void tearDown() throws Exception {
+		RepeatSynchronizationManager.clear();		
+		super.tearDown();
+	}
+
+	public void testInjection() throws Exception {
+		final TestBeanAware foo = new TestBeanAware();
+		Object value = scope.get("foo", new ObjectFactory() {
+			public Object getObject() throws BeansException {
+				return foo;
+			}
+		});
+		assertEquals(foo, value);
+		assertTrue(context.hasAttribute("foo"));
+		assertEquals(parameters, foo.getJobParameters());
+	}
+
+	public void testFailedInjection() throws Exception {
+		// Null JobInstance so no parameters
+		context.getStepExecution().getJobExecution().setJobInstance(null);
+		final TestBeanAware foo = new TestBeanAware();
+		Object value = scope.get("foo", new ObjectFactory() {
+			public Object getObject() throws BeansException {
+				return foo;
+			}
+		});
+		assertEquals(foo, value);
+		assertTrue(context.hasAttribute("foo"));
+		assertEquals(null, foo.getJobParameters());
+	}
+
+	public static class TestBeanAware implements JobParametersAware {
+		private JobParameters jobParameters;
+		public void setJobParameters(JobParameters jobParameters) {
+			this.jobParameters = jobParameters;
+		}
+		public JobParameters getJobParameters() {
+			return jobParameters;
+		}
+	}
+}

spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java

@@ -34,7 +34,6 @@ import org.springframework.batch.execution.repository.SimpleJobRepository;
 import org.springframework.batch.execution.repository.dao.MapJobExecutionDao;
 import org.springframework.batch.execution.repository.dao.MapJobInstanceDao;
 import org.springframework.batch.execution.repository.dao.MapStepExecutionDao;
-import org.springframework.batch.execution.scope.StepSynchronizationManager;
 import org.springframework.batch.execution.step.support.JobRepositorySupport;
 import org.springframework.batch.execution.step.support.StepInterruptionPolicy;
 import org.springframework.batch.io.exception.BatchCriticalException;
@@ -53,7 +52,6 @@ import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.batch.repeat.RepeatContext;
 import org.springframework.batch.repeat.exception.handler.DefaultExceptionHandler;
 import org.springframework.batch.repeat.exception.handler.ExceptionHandler;
-import org.springframework.batch.repeat.interceptor.RepeatListenerSupport;
 import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
 import org.springframework.batch.repeat.support.RepeatTemplate;
 import org.springframework.batch.support.PropertiesConverter;
@@ -143,56 +141,6 @@ public class ItemOrientedStepTests extends TestCase {
 
 	}
 
-	public void testStepContextInitialized() throws Exception {
-
-		template = new RepeatTemplate();
-
-		// Only process one item:
-		template.setCompletionPolicy(new SimpleCompletionPolicy(1));
-		itemOrientedStep.setChunkOperations(template);
-
-		final JobExecution jobExecution = new JobExecution(jobInstance);
-		final StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecution);
-
-		itemOrientedStep.setItemReader(new AbstractItemReader() {
-			public Object read() throws Exception {
-				assertEquals(itemOrientedStep.getName(), stepExecution.getStepName());
-				assertNotNull(StepSynchronizationManager.getContext().getStepExecution());
-				return "foo";
-			}
-		});
-
-		itemOrientedStep.execute(stepExecution);
-		assertEquals(1, processed.size());
-
-	}
-
-	public void testStepContextInitializedBeforeTasklet() throws Exception {
-
-		template = new RepeatTemplate();
-
-		// Only process one chunk:
-		template.setCompletionPolicy(new SimpleCompletionPolicy(1));
-		itemOrientedStep.setStepOperations(template);
-
-		final JobExecution jobExecution = new JobExecution(jobInstance);
-		jobExecution.setId(new Long(1));
-		final StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecution);
-
-		template.setListener(new RepeatListenerSupport() {
-			public void open(RepeatContext context) {
-				assertNotNull(StepSynchronizationManager.getContext().getStepExecution());
-				assertEquals(stepExecution, StepSynchronizationManager.getContext().getStepExecution());
-				// StepScope can obtain id information....
-				assertNotNull(StepSynchronizationManager.getContext().getIdentifier());
-			}
-		});
-
-		itemOrientedStep.execute(stepExecution);
-		assertEquals(1, processed.size());
-
-	}
-
 	public void testRepository() throws Exception {
 
 		SimpleJobRepository repository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(), new MapStepExecutionDao());
@@ -622,7 +570,6 @@ public class ItemOrientedStepTests extends TestCase {
 		private boolean restoreFromCalledWithSomeContext = false;
 
 		public Object read() throws Exception {
-			StepSynchronizationManager.getContext().setAttribute("TASKLET_TEST", this);
 			return "item";
 		}