OPEN - issue BATCH-282: Make input parameters easier to access from ItemReaders, etc.

Add StepScope and integrate with AbstractStep.  No support yet for late binding of step and job attributes.

spring-batch-core/pom.xml

@@ -69,6 +69,14 @@
 			<groupId>junit</groupId>
 			<artifactId>junit</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>aspectj</groupId>
+			<artifactId>aspectjrt</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>aspectj</groupId>
+			<artifactId>aspectjweaver</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>cglib</groupId>
 			<artifactId>cglib-nodep</artifactId>

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

@@ -0,0 +1,169 @@
+/*
+ * 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.core.scope;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.UnexpectedJobExecutionException;
+import org.springframework.batch.repeat.context.SynchronizedAttributeAccessor;
+import org.springframework.util.Assert;
+
+/**
+ * Simple implementation of {@link StepContext}.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class StepContext extends SynchronizedAttributeAccessor {
+
+	private StepExecution stepExecution;
+
+	private Map<String, Set<Runnable>> callbacks = new HashMap<String, Set<Runnable>>();
+
+	/**
+	 * Create a new instance of {@link StepContext} for this
+	 * {@link StepExecution}.
+	 * 
+	 * @param stepExecution a step execution
+	 */
+	public StepContext(StepExecution stepExecution) {
+		super();
+		Assert.notNull(stepExecution, "A StepContext must have a non-null StepExecution");
+		this.stepExecution = stepExecution;
+	}
+
+	/**
+	 * Allow clients to register callbacks for clean up on close.
+	 * 
+	 * @param name the callback id (unique attribute key in this context)
+	 * @param callback a callback to execute on close
+	 */
+	public void registerDestructionCallback(String name, Runnable callback) {
+		synchronized (callbacks) {
+			Set<Runnable> set = callbacks.get(name);
+			if (set == null) {
+				set = new HashSet<Runnable>();
+				callbacks.put(name, set);
+			}
+			set.add(callback);
+		}
+	}
+
+	private void unregisterDestructionCallbacks(String name) {
+		synchronized (callbacks) {
+			callbacks.remove(name);
+		}
+	}
+
+	/**
+	 * Override base class behaviour to ensure destruction callbacks are
+	 * unregistered as well as the default behaviour.
+	 * 
+	 * @see SynchronizedAttributeAccessor#removeAttribute(String)
+	 */
+	@Override
+	public Object removeAttribute(String name) {
+		unregisterDestructionCallbacks(name);
+		return super.removeAttribute(name);
+	}
+
+	/**
+	 * Clean up the context at the end of a step execution. Must be called once
+	 * at the end of a step execution to honour the destruction callback
+	 * contract from the {@link StepScope}.
+	 */
+	public void close() {
+
+		List<Exception> errors = new ArrayList<Exception>();
+
+		Map<String, Set<Runnable>> copy = Collections.unmodifiableMap(callbacks);
+
+		for (String key : copy.keySet()) {
+			Set<Runnable> set = copy.get(key);
+			for (Runnable callback : set) {
+				if (callback != null) {
+					/*
+					 * The documentation of the interface says that these
+					 * callbacks must not throw exceptions, but we don't trust
+					 * them necessarily...
+					 */
+					try {
+						callback.run();
+					}
+					catch (RuntimeException t) {
+						errors.add(t);
+					}
+				}
+			}
+		}
+
+		if (errors.isEmpty()) {
+			return;
+		}
+
+		Exception error = (Exception) errors.get(0);
+		if (error instanceof RuntimeException) {
+			throw (RuntimeException) error;
+		}
+		else {
+			throw new UnexpectedJobExecutionException("Could not close step context, rethrowing first of "
+					+ errors.size() + " execptions.", error);
+		}
+	}
+
+	public StepExecution getStepExecution() {
+		return stepExecution;
+	}
+
+	/**
+	 * Extend the base class method to include the step execution itself as a
+	 * key (i.e. two contexts are only equal if their step executions are the
+	 * same).
+	 * 
+	 * @see SynchronizedAttributeAccessor#equals(Object)
+	 */
+	@Override
+	public boolean equals(Object other) {
+		if (!(other instanceof StepContext))
+			return false;
+		if (other == this)
+			return true;
+		StepContext context = (StepContext) other;
+		if (context.stepExecution == stepExecution) {
+			return true;
+		}
+		return stepExecution.equals(context.stepExecution);
+	}
+
+	/**
+	 * Overrides the default behaviour to provide a hash code based only on the
+	 * step execution.
+	 * 
+	 * @see SynchronizedAttributeAccessor#hashCode()
+	 */
+	@Override
+	public int hashCode() {
+		return stepExecution.hashCode();
+	}
+
+}

spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepContextRepeatCallback.java

@@ -0,0 +1,66 @@
+/*
+ * 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.core.scope;
+
+import org.springframework.batch.core.Step;
+import org.springframework.batch.repeat.ExitStatus;
+import org.springframework.batch.repeat.RepeatCallback;
+import org.springframework.batch.repeat.RepeatContext;
+
+/**
+ * Convenient base class for clients who need to do something in a repeat
+ * callback inside a {@link Step}.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public abstract class StepContextRepeatCallback implements RepeatCallback {
+
+	private final StepContext stepContext;
+
+	/**
+	 * @param stepContext
+	 */
+	public StepContextRepeatCallback(StepContext stepContext) {
+		this.stepContext = stepContext;
+	}
+
+	/**
+	 * Manage the {@link StepContext} lifecycle to ensure that the current
+	 * thread has a reference to the context, even if the callback is executed
+	 * in a pooled thread.
+	 * 
+	 * @see RepeatCallback#doInIteration(RepeatContext)
+	 */
+	public ExitStatus doInIteration(RepeatContext context) throws Exception {
+		StepSynchronizationManager.register(stepContext);
+		try {
+			return doInStepContext(context, stepContext);
+		}
+		finally {
+			StepSynchronizationManager.close();
+		}
+	}
+
+	/**
+	 * @param context
+	 * @param stepContext
+	 * @return the exit status from the execution
+	 * @throws Exception
+	 */
+	public abstract ExitStatus doInStepContext(RepeatContext context, StepContext stepContext) throws Exception;
+
+}

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

@@ -0,0 +1,224 @@
+/*
+ * 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.core.scope;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.aop.scope.ScopedProxyUtils;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.ObjectFactory;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.config.BeanDefinitionHolder;
+import org.springframework.beans.factory.config.BeanDefinitionVisitor;
+import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.beans.factory.config.Scope;
+import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
+import org.springframework.beans.factory.support.BeanDefinitionRegistry;
+import org.springframework.core.Ordered;
+import org.springframework.util.Assert;
+import org.springframework.util.StringValueResolver;
+
+/**
+ * Scope for step context. Objects in this scope with <aop:scoped-proxy/>
+ * use the Spring container as an object factory, so there is only one instance
+ * of such a bean per executing step.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class StepScope implements Scope, BeanFactoryPostProcessor, Ordered {
+
+	private Log logger = LogFactory.getLog(getClass());
+
+	private int order = Ordered.LOWEST_PRECEDENCE;
+
+	private Object mutex = new Object();
+
+	/**
+	 * @param order the order value to set priority of callback execution for
+	 * the {@link BeanFactoryPostProcessor} part of this scope bean.
+	 */
+	public void setOrder(int order) {
+		this.order = order;
+	}
+
+	public int getOrder() {
+		return order;
+	}
+
+	/**
+	 * Context key for clients to use for conversation identifier.
+	 */
+	public static final String ID_KEY = "STEP_IDENTIFIER";
+
+	private String name = "step";
+
+	private boolean proxyTargetClass;
+
+	/**
+	 * @see Scope#get(String, ObjectFactory)
+	 */
+	public Object get(String name, ObjectFactory objectFactory) {
+
+		StepContext context = getContext();
+		Object scopedObject = context.getAttribute(name);
+
+		if (scopedObject == null) {
+
+			synchronized (mutex) {
+				scopedObject = context.getAttribute(name);
+				if (scopedObject == null) {
+
+					logger.debug(String.format("Creating object in scope=%s, name=%s", this.name, name));
+
+					/**
+					 * Here is where we need to inject some context (a root
+					 * object for expressions). The ObjectFactory could take a
+					 * parameter?
+					 */
+					scopedObject = objectFactory.getObject();
+					context.setAttribute(name, scopedObject);
+
+				}
+
+			}
+
+		}
+		return scopedObject;
+	}
+
+	/**
+	 * @see Scope#getConversationId()
+	 */
+	public String getConversationId() {
+		StepContext context = getContext();
+		Object id = context.getAttribute(ID_KEY);
+		return "" + id;
+	}
+
+	/**
+	 * @see Scope#registerDestructionCallback(String, Runnable)
+	 */
+	public void registerDestructionCallback(String name, Runnable callback) {
+		StepContext context = getContext();
+		logger.debug(String.format("Registered destruction callback in scope=%s, name=%s", this.name, name));
+		context.registerDestructionCallback(name, callback);
+	}
+
+	/**
+	 * @see Scope#remove(String)
+	 */
+	public Object remove(String name) {
+		StepContext context = getContext();
+		logger.debug(String.format("Removing from scope=%s, name=%s", this.name, name));
+		return context.removeAttribute(name);
+	}
+
+	/**
+	 * Get an attribute accessor in the form of a {@link StepContext} that can
+	 * be used to store scoped bean instances.
+	 * 
+	 * @return the current step context which we can use as a scope storage
+	 * medium
+	 */
+	private StepContext getContext() {
+		StepContext context = StepSynchronizationManager.getContext();
+		if (context == null) {
+			throw new IllegalStateException("No context holder available for step scope");
+		}
+		return context;
+	}
+
+	/**
+	 * Register this scope with the enclosing BeanFactory.
+	 * 
+	 * @see BeanFactoryPostProcessor#postProcessBeanFactory(ConfigurableListableBeanFactory)
+	 * 
+	 * @param beanFactory the BeanFactory to register with
+	 * @throws BeansException if there is a problem.
+	 */
+	public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
+		beanFactory.registerScope(name, this);
+		Assert.state(beanFactory instanceof BeanDefinitionRegistry,
+				"BeanFactory was not a BeanDefinitionRegistry, so StepScope cannot be used.");
+		Scopifier scopifier = new Scopifier((BeanDefinitionRegistry) beanFactory, name, proxyTargetClass);
+		for (String beanName : beanFactory.getBeanDefinitionNames()) {
+			BeanDefinition definition = beanFactory.getBeanDefinition(beanName);
+			// Replace this or any of its inner beans with scoped proxy if it
+			// has this scope
+			scopifier.visitBeanDefinition(definition);
+		}
+	}
+
+	/**
+	 * Public setter for the name property. This can then be used as a bean
+	 * definition attribute, e.g. scope="step". Defaults to "step".
+	 * 
+	 * @param name the name to set for this scope.
+	 */
+	public void setName(String name) {
+		this.name = name;
+	}
+
+	/**
+	 * Helper class to scan a bean definition hierarchy looking for scoped
+	 * objects and modifying their properties. In particular it forces the use
+	 * of auto-proxy for step scoped beans.
+	 * 
+	 * @author Dave Syer
+	 * 
+	 */
+	private static class Scopifier extends BeanDefinitionVisitor {
+
+		private final boolean proxyTargetClass;
+
+		private final BeanDefinitionRegistry registry;
+
+		private final String scope;
+
+		public Scopifier(BeanDefinitionRegistry registry, String scope, boolean proxyTargetClass) {
+			super(new StringValueResolver() {
+				public String resolveStringValue(String value) {
+					return value;
+				}
+			});
+			this.registry = registry;
+			this.proxyTargetClass = proxyTargetClass;
+			this.scope = scope;
+		}
+
+		@Override
+		protected Object resolveValue(Object value) {
+			if (value instanceof BeanDefinition) {
+				BeanDefinition definition = (BeanDefinition) value;
+				if (scope.equals(definition.getScope())) {
+					String beanName = BeanDefinitionReaderUtils.generateBeanName(definition, registry);
+					return ScopedProxyUtils.createScopedProxy(new BeanDefinitionHolder(definition, beanName), registry,
+							proxyTargetClass);
+				}
+			}
+			else if (value instanceof BeanDefinitionHolder) {
+				BeanDefinitionHolder definition = (BeanDefinitionHolder) value;
+				if (scope.equals(definition.getBeanDefinition().getScope())) {
+					return ScopedProxyUtils.createScopedProxy(definition, registry, proxyTargetClass);
+				}
+			}
+			return value;
+		}
+	}
+
+}

spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepScopeManager.java

@@ -0,0 +1,32 @@
+package org.springframework.batch.core.scope;
+
+import org.aspectj.lang.annotation.Around;
+import org.aspectj.lang.annotation.Aspect;
+import org.springframework.batch.core.JobInterruptedException;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.StepExecution;
+
+/**
+ * Convenient aspect to wrap a single threaded step execution, where the
+ * implementation of the {@link Step} is not step scope aware (i.e. not the ones
+ * provided by the framework).
+ * 
+ * @author Dave Syer
+ * 
+ */
+// TODO: bin this?
+@Aspect
+public class StepScopeManager {
+
+	@Around("execution(void org.springframework.batch.core.Step+.execute(*)) && target(step) && args(stepExecution)")
+	public void execute(Step step, StepExecution stepExecution) throws JobInterruptedException {
+		StepSynchronizationManager.register(new StepContext(stepExecution));
+		try {
+			step.execute(stepExecution);
+		}
+		finally {
+			StepSynchronizationManager.release();
+		}
+	}
+
+}

spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepSynchronizationManager.java

@@ -0,0 +1,108 @@
+/*
+ * 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.core.scope;
+
+import java.util.Stack;
+
+import org.springframework.batch.core.Step;
+
+/**
+ * Central convenience class for framework use in managing the step scope
+ * context. Generally only to be used by implementations of {@link Step}. N.B.
+ * it is the responsibility of every {@link Step} implementation to ensure that
+ * a {@link StepContext} is available on every thread that might be involved in
+ * a step execution, including worker threads from a pool.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class StepSynchronizationManager {
+
+	/**
+	 * Don't use InheritableThreadLocal because there are side effects if a step
+	 * is trying to run multiple child steps (e.g. with partitioning).
+	 */
+	private static final ThreadLocal<Stack<StepContext>> contextHolder = new ThreadLocal<Stack<StepContext>>();
+
+	/**
+	 * Getter for the current context if there is one, otherwise returns null.
+	 * 
+	 * @return the current {@link StepContext} or null if there is none (if one
+	 * has not been registered for this thread).
+	 */
+	public static StepContext getContext() {
+		Stack<StepContext> current = getCurrent();
+		if (current.isEmpty()) {
+			return null;
+		}
+		return current.peek();
+	}
+
+	/**
+	 * Method for registering a context with the current thread - always put a
+	 * matching {@link #close()} call in a finally block to ensure that the
+	 * correct context is available in the enclosing block.
+	 * 
+	 * @param context the step context to register
+	 */
+	public static void register(StepContext context) {
+		if (context == null) {
+			return;
+		}
+		getCurrent().push(context);
+	}
+
+	/**
+	 * Method for de-registering the current context - should always and only be
+	 * used by in conjunction with a matching {@link #register(StepContext)} to
+	 * ensure that {@link #getContext()} always returns the correct value. Does
+	 * not call {@link StepContext#close()} - that is left up to the caller
+	 * because he has a reference to the context (having registered it) and only
+	 * he has knowledge of when the step actually ended.
+	 */
+	public static void close() {
+		StepContext oldSession = getContext();
+		if (oldSession == null) {
+			return;
+		}
+		getCurrent().pop();
+	}
+
+	private static Stack<StepContext> getCurrent() {
+		if (contextHolder.get() == null) {
+			contextHolder.set(new Stack<StepContext>());
+		}
+		return contextHolder.get();
+	}
+
+	/**
+	 * A "deep" close operation. Call this instead of {@link #close()} if the
+	 * step execution for the current context is ending. Delegates to
+	 * {@link StepContext#close()} and then ensures that {@link #close()} is
+	 * also called in a finally block.
+	 */
+	public static void release() {
+		StepContext context = getContext();
+		try {
+			if (context!=null) {
+				context.close();
+			}
+		} finally {
+			close();
+		}
+	}
+
+}

spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java

@@ -31,6 +31,8 @@ import org.springframework.batch.core.launch.NoSuchJobException;
 import org.springframework.batch.core.launch.support.ExitCodeMapper;
 import org.springframework.batch.core.listener.CompositeStepExecutionListener;
 import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.scope.StepContext;
+import org.springframework.batch.core.scope.StepSynchronizationManager;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.beans.factory.BeanNameAware;
@@ -143,12 +145,12 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
 	/**
 	 * Extension point for subclasses to execute business logic. 
 	 * 
-	 * @param stepExecution
+	 * @param stepContext the current step context
 	 * @return {@link ExitStatus} to show whether the step is finished
 	 * processing.
 	 * @throws Exception
 	 */
-	protected abstract ExitStatus doExecute(StepExecution stepExecution) throws Exception;
+	protected abstract ExitStatus doExecute(StepContext stepContext) throws Exception;
 
 	/**
 	 * Extension point for subclasses to provide callbacks to their
@@ -175,7 +177,7 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
 	/**
 	 * Template method for step execution logic - calls abstract methods for
 	 * resource initialization ({@link #open(ExecutionContext)}), execution
-	 * logic ({@link #doExecute(StepExecution)}) and resource closing ({@link #close(ExecutionContext)}).
+	 * logic ({@link #doExecute(StepContext)}) and resource closing ({@link #close(ExecutionContext)}).
 	 */
 	public final void execute(StepExecution stepExecution) throws JobInterruptedException,
 			UnexpectedJobExecutionException {
@@ -186,11 +188,14 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
 		ExitStatus exitStatus = ExitStatus.FAILED;
 		Exception commitException = null;
 
+		StepContext stepContext = new StepContext(stepExecution);
+		StepSynchronizationManager.register(stepContext);
+
 		try {
 			getCompositeListener().beforeStep(stepExecution);
 			open(stepExecution.getExecutionContext());
 
-			exitStatus = doExecute(stepExecution);
+			exitStatus = doExecute(stepContext);
 
 			// Check if someone is trying to stop us
 			if (stepExecution.isTerminateOnly()) {
@@ -256,6 +261,8 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
 				logger.error("Exception while closing step execution resources", e);
 				stepExecution.addFailureException(e);
 			}
+			
+			StepSynchronizationManager.release();
 
 			if (commitException != null) {
 				stepExecution.setStatus(BatchStatus.UNKNOWN);

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

@@ -26,6 +26,8 @@ import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.StepExecutionListener;
 import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.scope.StepContext;
+import org.springframework.batch.core.scope.StepContextRepeatCallback;
 import org.springframework.batch.core.step.AbstractStep;
 import org.springframework.batch.core.step.StepExecutionSynchronizer;
 import org.springframework.batch.core.step.StepExecutionSynchronizerFactory;
@@ -37,7 +39,6 @@ import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.item.support.CompositeItemStream;
 import org.springframework.batch.repeat.ExitStatus;
-import org.springframework.batch.repeat.RepeatCallback;
 import org.springframework.batch.repeat.RepeatContext;
 import org.springframework.batch.repeat.RepeatOperations;
 import org.springframework.batch.repeat.support.RepeatTemplate;
@@ -225,16 +226,19 @@ public class TaskletStep extends AbstractStep {
 	 * execution
 	 * 
 	 */
-	protected ExitStatus doExecute(final StepExecution stepExecution) throws Exception {
+	@Override
+	protected ExitStatus doExecute(StepContext stepContext) throws Exception {
+		StepExecution stepExecution = stepContext.getStepExecution();
 		stream.update(stepExecution.getExecutionContext());
 		getJobRepository().updateExecutionContext(stepExecution);
 
-		return stepOperations.iterate(new RepeatCallback() {
+		return stepOperations.iterate(new StepContextRepeatCallback(stepContext ) {
 
 			final Queue<AttributeAccessor> attributeQueue = new LinkedBlockingQueue<AttributeAccessor>();
 
-			public ExitStatus doInIteration(RepeatContext context) throws Exception {
+			public ExitStatus doInStepContext(RepeatContext context, StepContext stepContext) throws Exception {
 
+				StepExecution stepExecution = stepContext.getStepExecution();
 				ExceptionHolder fatalException = new ExceptionHolder();
 
 				StepContribution contribution = stepExecution.createStepContribution();

spring-batch-core/src/test/java/org/springframework/batch/core/scope/Collaborator.java

@@ -0,0 +1,7 @@
+package org.springframework.batch.core.scope;
+
+public interface Collaborator {
+
+	public abstract String getName();
+
+}

spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepContextRepeatCallbackTests.java

@@ -0,0 +1,45 @@
+/*
+ * 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.core.scope;
+
+import static org.junit.Assert.assertEquals;
+
+import org.junit.Test;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.repeat.ExitStatus;
+import org.springframework.batch.repeat.RepeatContext;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class StepContextRepeatCallbackTests {
+
+	@Test
+	public void testDoInIteration() throws Exception {
+		 StepContext stepContext = new StepContext(new StepExecution("foo", new JobExecution(0L), 123L));
+		StepContextRepeatCallback callback = new  StepContextRepeatCallback(stepContext ) {
+			@Override
+			public ExitStatus doInStepContext(RepeatContext context, StepContext stepContext) throws Exception {
+				assertEquals(Long.valueOf(123), stepContext.getStepExecution().getId());
+				return ExitStatus.NOOP;
+			}
+		 };
+		 assertEquals(ExitStatus.NOOP, callback.doInIteration(null));
+	}
+
+}

spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepContextTests.java

@@ -0,0 +1,131 @@
+/*
+ * 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.core.scope;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Test;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.StepExecution;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class StepContextTests {
+
+	private List<String> list = new ArrayList<String>();
+
+	private StepExecution stepExecution = new StepExecution("step", new JobExecution(0L));
+
+	private StepContext context = new StepContext(stepExecution);
+
+	@Test
+	public void testGetStepExecution() {
+		context = new StepContext(stepExecution);
+		assertNotNull(context.getStepExecution());
+	}
+
+	@Test
+	public void testNullStepExecution() {
+		try {
+			context = new StepContext(null);
+			fail("Expected IllegalArgumentException");
+		}
+		catch (IllegalArgumentException e) {
+			// expected
+		}
+	}
+
+	@Test
+	public void testEqualsSelf() {
+		assertEquals(context, context);
+	}
+
+	@Test
+	public void testNotEqualsNull() {
+		assertFalse(context.equals(null));
+	}
+
+	@Test
+	public void testEqualsContextWithSameStepExecution() {
+		assertEquals(new StepContext(stepExecution), context);
+	}
+
+	@Test
+	public void testDestructionCallbackSunnyDay() throws Exception {
+		context.setAttribute("foo", "FOO");
+		context.registerDestructionCallback("foo", new Runnable() {
+			public void run() {
+				list.add("bar");
+			}
+		});
+		context.close();
+		assertEquals(1, list.size());
+		assertEquals("bar", list.get(0));
+	}
+
+	@Test
+	public void testDestructionCallbackMissingAttribute() throws Exception {
+		context.registerDestructionCallback("foo", new Runnable() {
+			public void run() {
+				list.add("bar");
+			}
+		});
+		context.close();
+		// Yes the callback should be called even if the attribute is missing -
+		// for inner beans
+		assertEquals(1, list.size());
+	}
+
+	@Test
+	public void testDestructionCallbackWithException() throws Exception {
+		context.setAttribute("foo", "FOO");
+		context.setAttribute("bar", "BAR");
+		context.registerDestructionCallback("bar", new Runnable() {
+			public void run() {
+				list.add("spam");
+				throw new RuntimeException("fail!");
+			}
+		});
+		context.registerDestructionCallback("foo", new Runnable() {
+			public void run() {
+				list.add("bar");
+				throw new RuntimeException("fail!");
+			}
+		});
+		try {
+			context.close();
+			fail("Expected RuntimeException");
+		}
+		catch (RuntimeException e) {
+			// We don't care which one was thrown...
+			assertEquals("fail!", e.getMessage());
+		}
+		// ...but we do care that both were executed:
+		assertEquals(2, list.size());
+		assertTrue(list.contains("bar"));
+		assertTrue(list.contains("spam"));
+	}
+
+}

spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepScopeIntegrationTests.java

@@ -0,0 +1,73 @@
+package org.springframework.batch.core.scope;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class StepScopeIntegrationTests {
+	
+	@Autowired
+	@Qualifier("vanilla")
+	private Step vanilla;
+	
+	@Autowired
+	@Qualifier("proxied")
+	private Step proxied;
+
+	@Autowired
+	@Qualifier("enhanced")
+	private Step enhanced;
+
+	@Before
+	@After
+	public void start() {
+		TestStep.reset();
+	}
+		
+	@Test
+	public void testScopeCreation() throws Exception {
+		vanilla.execute(new StepExecution("foo",new JobExecution(11L)));
+		assertNotNull(TestStep.getContext());
+		assertNull(StepSynchronizationManager.getContext());
+	}
+
+	@Test
+	public void testScopedProxy() throws Exception {
+		proxied.execute(new StepExecution("foo",new JobExecution(11L)));
+		assertTrue(TestStep.getContext().attributeNames().length>0);
+		String collaborator = (String) TestStep.getContext().getAttribute("collaborator");
+		assertNotNull(collaborator);
+		assertEquals("bar", collaborator);
+	}
+
+	@Test
+	public void testExecutionContext() throws Exception {
+		StepExecution stepExecution = new StepExecution("foo",new JobExecution(11L));
+		ExecutionContext executionContext = new ExecutionContext();
+		executionContext.put("name", "spam");
+		stepExecution.setExecutionContext(executionContext);
+		enhanced.execute(stepExecution);
+		assertTrue(TestStep.getContext().attributeNames().length>0);
+		String collaborator = (String) TestStep.getContext().getAttribute("collaborator");
+		assertNotNull(collaborator);
+		assertEquals("bar", collaborator);
+	}
+
+}

spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepScopeTests.java

@@ -0,0 +1,188 @@
+/*
+ * 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.core.scope;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.ObjectFactory;
+import org.springframework.context.support.StaticApplicationContext;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class StepScopeTests {
+
+	private StepScope scope = new StepScope();
+
+	private StepContext context = new StepContext(new StepExecution("foo", new JobExecution(0L)));
+
+	@Before
+	public void setUp() throws Exception {
+		StepSynchronizationManager.register(context);
+	}
+
+	@After
+	public void tearDown() throws Exception {
+		StepSynchronizationManager.close();
+	}
+
+	@Test
+	public void testGetWithNoContext() throws Exception {
+		final String foo = "bar";
+		StepSynchronizationManager.close();
+		try {
+			scope.get("foo", new ObjectFactory() {
+				public Object getObject() throws BeansException {
+					return foo;
+				}
+			});
+			fail("Expected IllegalStateException");
+		}
+		catch (IllegalStateException e) {
+			// expected
+		}
+
+	}
+
+	@Test
+	public void testGetWithNothingAlreadyThere() {
+		final String foo = "bar";
+		Object value = scope.get("foo", new ObjectFactory() {
+			public Object getObject() throws BeansException {
+				return foo;
+			}
+		});
+		assertEquals(foo, value);
+		assertTrue(context.hasAttribute("foo"));
+	}
+
+	@Test
+	public void testGetWithSomethingAlreadyThere() {
+		context.setAttribute("foo", "bar");
+		Object value = scope.get("foo", new ObjectFactory() {
+			public Object getObject() throws BeansException {
+				return null;
+			}
+		});
+		assertEquals("bar", value);
+		assertTrue(context.hasAttribute("foo"));
+	}
+
+	@Test
+	public void testGetWithSomethingAlreadyInParentContext() {
+		context.setAttribute("foo", "bar");
+		StepContext context = new StepContext(new StepExecution("bar", new JobExecution(0L)));
+		StepSynchronizationManager.register(context);
+		Object value = scope.get("foo", new ObjectFactory() {
+			public Object getObject() throws BeansException {
+				return "spam";
+			}
+		});
+		assertEquals("spam", value);
+		assertTrue(context.hasAttribute("foo"));
+		StepSynchronizationManager.close();
+		assertEquals("bar", scope.get("foo", null));
+	}
+
+	@Test
+	public void testGetConversationId() {
+		String id = scope.getConversationId();
+		assertNotNull(id);
+	}
+
+	@Test
+	public void testGetConversationIdFromAttribute() {
+		context.setAttribute(StepScope.ID_KEY, "foo");
+		String id = scope.getConversationId();
+		assertEquals("foo", id);
+	}
+
+	@Test
+	public void testRegisterDestructionCallback() {
+		final List<String> list = new ArrayList<String>();
+		context.setAttribute("foo", "bar");
+		scope.registerDestructionCallback("foo", new Runnable() {
+			public void run() {
+				list.add("foo");
+			}
+		});
+		assertEquals(0, list.size());
+		// When the context is closed, provided the attribute exists the
+		// callback is called...
+		context.close();
+		assertEquals(1, list.size());
+	}
+
+	@Test
+	public void testRegisterAnotherDestructionCallback() {
+		final List<String> list = new ArrayList<String>();
+		context.setAttribute("foo", "bar");
+		scope.registerDestructionCallback("foo", new Runnable() {
+			public void run() {
+				list.add("foo");
+			}
+		});
+		scope.registerDestructionCallback("foo", new Runnable() {
+			public void run() {
+				list.add("bar");
+			}
+		});
+		assertEquals(0, list.size());
+		// When the context is closed, provided the attribute exists the
+		// callback is called...
+		context.close();
+		assertEquals(2, list.size());
+	}
+
+	@Test
+	public void testRemove() {
+		context.setAttribute("foo", "bar");
+		scope.remove("foo");
+		assertFalse(context.hasAttribute("foo"));
+	}
+
+	@Test
+	public void testOrder() throws Exception {
+		assertEquals(Integer.MAX_VALUE, scope.getOrder());
+		scope.setOrder(11);
+		assertEquals(11, scope.getOrder());
+	}
+
+	@Test
+	public void testName() throws Exception {
+		scope.setName("foo");
+		StaticApplicationContext beanFactory = new StaticApplicationContext();
+		scope.postProcessBeanFactory(beanFactory.getDefaultListableBeanFactory());
+		String[] scopes = beanFactory.getDefaultListableBeanFactory().getRegisteredScopeNames();
+		assertEquals(1, scopes.length);
+		assertEquals("foo", scopes[0]);
+	}
+
+}

spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepSynchronizationManagerTests.java

@@ -0,0 +1,87 @@
+package org.springframework.batch.core.scope;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.StepExecution;
+
+public class StepSynchronizationManagerTests {
+
+	private StepExecution stepExecution = new StepExecution("step", new JobExecution(0L));
+
+	@Before
+	@After
+	public void start() {
+		while (StepSynchronizationManager.getContext() != null) {
+			StepSynchronizationManager.close();
+		}
+	}
+
+	@Test
+	public void testGetContext() {
+		assertNull(StepSynchronizationManager.getContext());
+		StepSynchronizationManager.register(new StepContext(stepExecution));
+		assertNotNull(StepSynchronizationManager.getContext());
+	}
+
+	@Test
+	public void testClose() {
+		StepContext context = new StepContext(stepExecution);
+		final List<String> list = new ArrayList<String>();
+		context.registerDestructionCallback("foo", new Runnable() {
+			public void run() {
+				list.add("foo");
+			}
+		});
+		StepSynchronizationManager.register(context);
+		StepSynchronizationManager.close();
+		assertNull(StepSynchronizationManager.getContext());
+		assertEquals(0, list.size());
+	}
+
+	@Test
+	public void testRelease() {
+		StepContext context = new StepContext(stepExecution);
+		final List<String> list = new ArrayList<String>();
+		context.registerDestructionCallback("foo", new Runnable() {
+			public void run() {
+				list.add("foo");
+			}
+		});
+		StepSynchronizationManager.register(context);
+		// On release we expect the destruction callbacks to be called
+		StepSynchronizationManager.release();
+		assertNull(StepSynchronizationManager.getContext());
+		assertEquals(1, list.size());
+	}
+
+	@Test
+	public void testRegisterNull() {
+		assertNull(StepSynchronizationManager.getContext());
+		StepSynchronizationManager.register(null);
+		assertNull(StepSynchronizationManager.getContext());
+	}
+
+	@Test
+	public void testRegisterTwice() {
+		StepContext context = new StepContext(stepExecution);
+		StepSynchronizationManager.register(context);
+		StepSynchronizationManager.register(context);
+		StepSynchronizationManager.close();
+		// if someone registers you have to assume they are going to close, so
+		// the last thing you want is for the close to remove another context
+		// that someone else has registered
+		assertNotNull(StepSynchronizationManager.getContext());
+		StepSynchronizationManager.close();
+		assertNull(StepSynchronizationManager.getContext());
+	}
+
+}

spring-batch-core/src/test/java/org/springframework/batch/core/scope/TestCollaborator.java

@@ -0,0 +1,19 @@
+package org.springframework.batch.core.scope;
+
+
+public class TestCollaborator implements Collaborator {
+
+	private String name;
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.core.scope.Collaborator#getName()
+	 */
+	public String getName() {
+		return name;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+
+}

spring-batch-core/src/test/java/org/springframework/batch/core/scope/TestStep.java

@@ -0,0 +1,42 @@
+package org.springframework.batch.core.scope;
+
+import org.springframework.batch.core.JobInterruptedException;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.StepExecution;
+
+public class TestStep implements Step {
+
+	private static StepContext context;
+	
+	private Collaborator collaborator;
+	
+	public void setCollaborator(Collaborator collaborator) {
+		this.collaborator = collaborator;
+	}
+
+	public static StepContext getContext() {
+		return context;
+	}
+
+	public static void reset() {
+		context = null;
+	}
+
+	public void execute(StepExecution stepExecution) throws JobInterruptedException {
+		context = StepSynchronizationManager.getContext();
+		context.setAttribute("collaborator", collaborator.getName());
+	}
+
+	public String getName() {
+		return "foo";
+	}
+
+	public int getStartLimit() {
+		return Integer.MAX_VALUE;
+	}
+
+	public boolean isAllowStartIfComplete() {
+		return false;
+	}
+
+}

spring-batch-core/src/test/java/org/springframework/batch/core/step/AbstractStepTests.java

@@ -16,6 +16,7 @@ import org.springframework.batch.core.JobInstance;
 import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.StepExecutionListener;
+import org.springframework.batch.core.scope.StepContext;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.util.Assert;
@@ -54,8 +55,8 @@ public class AbstractStepTests {
 			events.add("open");
 		}
 
-		protected ExitStatus doExecute(StepExecution stepExecution) throws Exception {
-			assertSame(execution, stepExecution);
+		protected ExitStatus doExecute(StepContext context) throws Exception {
+			assertSame(execution, context.getStepExecution());
 			events.add("doExecute");
 			return ExitStatus.FINISHED;
 		}
@@ -165,8 +166,9 @@ public class AbstractStepTests {
 	@Test
 	public void testFailure() throws Exception {
 		tested = new EventTrackingStep() {
-			protected ExitStatus doExecute(StepExecution stepExecution) throws Exception {
-				super.doExecute(stepExecution);
+			@Override
+			protected ExitStatus doExecute(StepContext context) throws Exception {
+				super.doExecute(context);
 				throw new RuntimeException("crash!");
 			}
 		};
@@ -202,9 +204,10 @@ public class AbstractStepTests {
 	@Test
 	public void testStoppedStep() throws Exception {
 		tested = new EventTrackingStep() {
-			protected ExitStatus doExecute(StepExecution stepExecution) throws Exception {
-				stepExecution.setTerminateOnly();
-				return super.doExecute(stepExecution);
+			@Override
+			protected ExitStatus doExecute(StepContext context) throws Exception {
+				context.getStepExecution().setTerminateOnly();
+				return super.doExecute(context);
 			}
 		};
 		tested.setJobRepository(repository);
@@ -237,8 +240,9 @@ public class AbstractStepTests {
 	@Test
 	public void testFailureInSavingExecutionContext() throws Exception {
 		tested = new EventTrackingStep() {
-			protected ExitStatus doExecute(StepExecution stepExecution) throws Exception {
-				super.doExecute(stepExecution);
+			@Override
+			protected ExitStatus doExecute(StepContext context) throws Exception {
+				super.doExecute(context);
 				return ExitStatus.FINISHED;
 			}
 		};

spring-batch-core/src/test/resources/org/springframework/batch/core/scope/StepScopeIntegrationTests-context.xml

@@ -0,0 +1,35 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
+	xmlns:p="http://www.springframework.org/schema/p" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd
+		http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd">
+	<bean id="vanilla" class="org.springframework.batch.core.scope.TestStep">
+		<property name="collaborator">
+			<bean class="org.springframework.batch.core.scope.TestCollaborator">
+				<property name="name" value="bar" />
+			</bean>
+		</property>
+	</bean>
+	<bean id="proxied" class="org.springframework.batch.core.scope.TestStep">
+		<property name="collaborator">
+			<bean class="org.springframework.batch.core.scope.TestCollaborator"
+				scope="step">
+				<property name="name" value="bar" />
+			</bean>
+		</property>
+	</bean>
+	<bean id="enhanced" class="org.springframework.batch.core.scope.TestStep">
+		<property name="collaborator">
+			<bean class="org.springframework.batch.core.scope.TestCollaborator"
+				scope="step">
+				<property name="name" value="bar" />
+			</bean>
+		</property>
+	</bean>
+	<aop:aspectj-autoproxy />
+	<bean class="org.springframework.batch.core.scope.StepScopeManager" />
+	<bean class="org.springframework.batch.core.scope.StepScope" />
+</beans>

spring-batch-integration/src/main/java/org/springframework/batch/integration/job/MessageOrientedStep.java

@@ -19,6 +19,7 @@ import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.JobInterruptedException;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.UnexpectedJobExecutionException;
+import org.springframework.batch.core.scope.StepContext;
 import org.springframework.batch.core.step.AbstractStep;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.repeat.ExitStatus;
@@ -93,15 +94,15 @@ public class MessageOrientedStep extends AbstractStep {
 		this.source = source;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.batch.core.step.AbstractStep#execute(org.springframework.batch.core.StepExecution)
+	/**
+	 * @see AbstractStep#execute(StepExecution)
 	 */
 	@Override
-	public ExitStatus doExecute(StepExecution stepExecution) throws JobInterruptedException,
+	public ExitStatus doExecute(StepContext stepContext) throws JobInterruptedException,
 			UnexpectedJobExecutionException {
 
-		JobExecutionRequest request = new JobExecutionRequest(stepExecution.getJobExecution());
+		StepExecution stepExecution = stepContext.getStepExecution();
+		JobExecutionRequest request = new JobExecutionRequest(stepExecution .getJobExecution());
 
 		ExecutionContext executionContext = stepExecution.getExecutionContext();
 

spring-batch-integration/src/main/java/org/springframework/batch/integration/job/StepExecutionMessageHandler.java

@@ -25,12 +25,14 @@ import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.beans.factory.annotation.Required;
+import org.springframework.integration.annotation.MessageEndpoint;
 import org.springframework.integration.annotation.ServiceActivator;
 
 /**
  * @author Dave Syer
  * 
  */
+@MessageEndpoint
 public class StepExecutionMessageHandler {
 
 	private Step step;