diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemProcessor.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemProcessor.java
index 87a5ee7cf..a90af2f51 100644
--- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemProcessor.java
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemProcessor.java
@@ -1,9 +1,27 @@
+/*
+ * 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.integration.async;
 
 import java.util.concurrent.Callable;
 import java.util.concurrent.Future;
 import java.util.concurrent.FutureTask;
 
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.scope.context.StepContext;
+import org.springframework.batch.core.scope.context.StepSynchronizationManager;
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.core.task.SyncTaskExecutor;
@@ -65,13 +83,36 @@ public class AsyncItemProcessor<I, O> implements ItemProcessor<I, Future<O>>, In
 	 * @see ItemProcessor#process(Object)
 	 */
 	public Future<O> process(final I item) throws Exception {
+		final StepExecution stepExecution = getStepExecution();
 		FutureTask<O> task = new FutureTask<O>(new Callable<O>() {
 			public O call() throws Exception {
-				return delegate.process(item);
+				if (stepExecution != null) {
+					StepSynchronizationManager.register(stepExecution);
+				}
+				try {
+					return delegate.process(item);
+				}
+				finally {
+					if (stepExecution != null) {
+						StepSynchronizationManager.close();
+					}
+				}
 			}
 		});
 		taskExecutor.execute(task);
 		return task;
 	}
 
+	/**
+	 * @return the current step execution if there is one
+	 */
+	private StepExecution getStepExecution() {
+		StepContext context = StepSynchronizationManager.getContext();
+		if (context==null) {
+			return null;
+		}
+		StepExecution stepExecution = context.getStepExecution();
+		return stepExecution;
+	}
+
 }
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java
index af5ea416b..97c69c890 100644
--- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java
@@ -1,3 +1,18 @@
+/*
+ * 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.integration.async;
 
 import java.util.ArrayList;
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/StepExecutionInterceptor.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/StepExecutionInterceptor.java
new file mode 100644
index 000000000..7fd5a9083
--- /dev/null
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/StepExecutionInterceptor.java
@@ -0,0 +1,53 @@
+/*
+ * 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.integration.async;
+
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.scope.context.StepContext;
+import org.springframework.batch.core.scope.context.StepSynchronizationManager;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.channel.ChannelInterceptor;
+import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter;
+import org.springframework.integration.support.MessageBuilder;
+
+/**
+ * A {@link ChannelInterceptor} that adds the current {@link StepExecution} (if
+ * there is one) as a header to the message. Downstream asynchronous handlers
+ * can then take advantage of the step context without needing to be step
+ * scoped, which is a problem for handlers executing in another thread because
+ * the scope context is not available.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class StepExecutionInterceptor extends ChannelInterceptorAdapter {
+
+	/**
+	 * The name of the header
+	 */
+	public static final String STEP_EXECUTION = "stepExecution";
+
+	@Override
+	public Message<?> preSend(Message<?> message, MessageChannel channel) {
+		StepContext context = StepSynchronizationManager.getContext();
+		if (context == null) {
+			return message;
+		}
+		return MessageBuilder.fromMessage(message).setHeader(STEP_EXECUTION, context.getStepExecution()).build();
+	}
+
+}
diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests.java
index 63c253dca..576645000 100644
--- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests.java
+++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests.java
@@ -1,3 +1,18 @@
+/*
+ * 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.integration.async;
 
 import static org.junit.Assert.assertNotNull;
@@ -5,11 +20,20 @@ import static org.junit.Assert.assertTrue;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.Callable;
 import java.util.concurrent.Future;
 
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.MethodRule;
 import org.junit.runner.RunWith;
+import org.junit.runners.model.FrameworkMethod;
+import org.junit.runners.model.Statement;
+import org.springframework.batch.core.JobParametersBuilder;
+import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.item.ItemProcessor;
+import org.springframework.batch.test.MetaDataInstanceFactory;
+import org.springframework.batch.test.StepScopeTestUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.integration.annotation.MessageEndpoint;
@@ -23,6 +47,32 @@ public class AsyncItemProcessorMessagingGatewayTests {
 
 	private AsyncItemProcessor<String, String> processor = new AsyncItemProcessor<String, String>();
 
+	private StepExecution stepExecution = MetaDataInstanceFactory.createStepExecution(new JobParametersBuilder().addLong("factor", 2L).toJobParameters());;
+
+	@Rule
+	public MethodRule rule = new MethodRule() {
+		public Statement apply(final Statement base, FrameworkMethod method, Object target) {
+			return new Statement() {
+				public void evaluate() throws Throwable {
+					StepScopeTestUtils.doInStepScope(stepExecution, new Callable<Void>() {
+						public Void call() throws Exception {
+							try {
+								base.evaluate();
+							}
+							catch (Exception e) {
+								throw e;
+							}
+							catch (Throwable e) {
+								throw new Error(e);
+							}
+							return null;
+						}
+					});
+				};
+			};
+		}
+	};
+
 	@Autowired
 	private ItemProcessor<String, String> delegate;
 
@@ -50,9 +100,18 @@ public class AsyncItemProcessorMessagingGatewayTests {
 
 	@MessageEndpoint
 	public static class Doubler {
+		private int factor = 1;
+
+		public void setFactor(int factor) {
+			this.factor = factor;
+		}
+
 		@ServiceActivator
 		public String cat(String value) {
-			return value + value;
+			for (int i=1; i<factor; i++) {
+				value += value;
+			}
+			return value;
 		}
 	}
 
diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorTests.java
index c6093d310..9636d870a 100644
--- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorTests.java
+++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorTests.java
@@ -1,3 +1,18 @@
+/*
+ * 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.integration.async;
 
 import static org.junit.Assert.assertEquals;
@@ -5,10 +20,15 @@ import static org.junit.Assert.assertTrue;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.Callable;
 import java.util.concurrent.Future;
 
 import org.junit.Test;
+import org.springframework.batch.core.scope.context.StepContext;
+import org.springframework.batch.core.scope.context.StepSynchronizationManager;
 import org.springframework.batch.item.ItemProcessor;
+import org.springframework.batch.test.MetaDataInstanceFactory;
+import org.springframework.batch.test.StepScopeTestUtils;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 
 public class AsyncItemProcessorTests {
@@ -33,6 +53,24 @@ public class AsyncItemProcessorTests {
 		assertEquals("foofoo", result.get());
 	}
 
+	@Test
+	public void testExecutionInStepScope() throws Exception {
+		delegate = new ItemProcessor<String, String>() {
+			public String process(String item) throws Exception {
+				StepContext context = StepSynchronizationManager.getContext();
+				assertTrue(context != null && context.getStepExecution() != null);
+				return item + item;
+			};
+		};
+		processor.setDelegate(delegate);
+		Future<String> result = StepScopeTestUtils.doInStepScope(MetaDataInstanceFactory.createStepExecution(), new Callable<Future<String>>() {
+			public Future<String> call() throws Exception {
+				return processor.process("foo");
+			}
+		});
+		assertEquals("foofoo", result.get());
+	}
+
 	@Test
 	public void testMultiExecution() throws Exception {
 		processor.setDelegate(delegate);
diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/PollingAsyncItemProcessorMessagingGatewayTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/PollingAsyncItemProcessorMessagingGatewayTests.java
new file mode 100644
index 000000000..271469926
--- /dev/null
+++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/PollingAsyncItemProcessorMessagingGatewayTests.java
@@ -0,0 +1,116 @@
+/*
+ * 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.integration.async;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.MethodRule;
+import org.junit.runner.RunWith;
+import org.junit.runners.model.FrameworkMethod;
+import org.junit.runners.model.Statement;
+import org.springframework.batch.core.JobParametersBuilder;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.item.ItemProcessor;
+import org.springframework.batch.test.MetaDataInstanceFactory;
+import org.springframework.batch.test.StepScopeTestUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.integration.annotation.Header;
+import org.springframework.integration.annotation.MessageEndpoint;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration
+public class PollingAsyncItemProcessorMessagingGatewayTests {
+
+	private AsyncItemProcessor<String, String> processor = new AsyncItemProcessor<String, String>();
+
+	private StepExecution stepExecution = MetaDataInstanceFactory.createStepExecution(new JobParametersBuilder().addLong("factor", 2L).toJobParameters());;
+
+	@Rule
+	public MethodRule rule = new MethodRule() {
+		public Statement apply(final Statement base, FrameworkMethod method, Object target) {
+			return new Statement() {
+				public void evaluate() throws Throwable {
+					StepScopeTestUtils.doInStepScope(stepExecution, new Callable<Void>() {
+						public Void call() throws Exception {
+							try {
+								base.evaluate();
+							}
+							catch (Exception e) {
+								throw e;
+							}
+							catch (Throwable e) {
+								throw new Error(e);
+							}
+							return null;
+						}
+					});
+				};
+			};
+		}
+	};
+
+	@Autowired
+	private ItemProcessor<String, String> delegate;
+
+	@Test
+	public void testMultiExecution() throws Exception {
+		processor.setDelegate(delegate);
+		processor.setTaskExecutor(new SimpleAsyncTaskExecutor());
+		List<Future<String>> list = new ArrayList<Future<String>>();
+		for (int count = 0; count < 10; count++) {
+			list.add(processor.process("foo" + count));
+		}
+		for (Future<String> future : list) {
+			String value = future.get();
+			/**
+			 * This delegate is a Spring Integration MessagingGateway. It can
+			 * easily return null because of a timeout, but that will be treated
+			 * by Batch as a filtered item, whereas it is really more like a
+			 * skip. So we have to throw an exception in the processor if an
+			 * unexpected null value comes back.
+			 */
+			assertNotNull(value);
+			assertTrue(value.matches("foo.*foo.*"));
+		}
+	}
+
+	@MessageEndpoint
+	public static class Doubler {
+
+		@ServiceActivator
+		public String cat(String value, @Header(value="stepExecution.jobExecution.jobInstance.jobParameters.getLong('factor')", required=false) Integer input) {
+			long factor = input==null ? 1 : input;
+			for (int i=1; i<factor; i++) {
+				value += value;
+			}
+			return value;
+		}
+
+	}
+
+}
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests-context.xml
index 9139dcf0e..5372b4efa 100644
--- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests-context.xml
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests-context.xml
@@ -1,25 +1,23 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<beans:beans xmlns="http://www.springframework.org/schema/integration"
-	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans"
-	xmlns:context="http://www.springframework.org/schema/context" xmlns:tx="http://www.springframework.org/schema/tx"
-	xsi:schemaLocation="http://www.springframework.org/schema/beans
-                                 http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
-                                 http://www.springframework.org/schema/integration
-                                 http://www.springframework.org/schema/integration/spring-integration-1.0.xsd
-                                 http://www.springframework.org/schema/context
-                                 http://www.springframework.org/schema/context/spring-context-2.5.xsd
-                                 http://www.springframework.org/schema/tx
-                                 http://www.springframework.org/schema/tx/spring-tx-2.5.xsd">
-	<annotation-config />
-	<channel id="requests" />
-	<channel id="replies">
-		<queue />
-	</channel>
-	<gateway service-interface="org.springframework.batch.item.ItemProcessor"
-		id="processor" default-reply-timeout="1000" default-request-channel="requests"
-		default-reply-channel="replies" />
-	<service-activator input-channel="requests"
-		output-channel="replies" ref="doubler" />
-	<beans:bean id="doubler"
-		class="org.springframework.batch.integration.async.AsyncItemProcessorMessagingGatewayTests$Doubler" />
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+                                 http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+                                 http://www.springframework.org/schema/integration
+                                 http://www.springframework.org/schema/integration/spring-integration-2.0.xsd">
+	<annotation-config />
+	<channel id="requests"/>
+	<channel id="replies">
+		<queue />
+	</channel>
+	<gateway service-interface="org.springframework.batch.item.ItemProcessor" id="processor" default-reply-timeout="1000"
+		default-request-channel="requests" default-reply-channel="replies" />
+	<service-activator input-channel="requests" output-channel="replies" ref="doubler"/>
+	<beans:bean id="doubler" class="org.springframework.batch.integration.async.AsyncItemProcessorMessagingGatewayTests$Doubler"
+		scope="step">
+		<beans:property name="factor" value="#{jobParameters['factor']}" />
+	</beans:bean>
+	<beans:bean class="org.springframework.batch.core.scope.StepScope">
+		<beans:property name="proxyTargetClass" value="true"/>
+	</beans:bean>
 </beans:beans>
\ No newline at end of file
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/async/PollingAsyncItemProcessorMessagingGatewayTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/async/PollingAsyncItemProcessorMessagingGatewayTests-context.xml
new file mode 100644
index 000000000..42df2c0d2
--- /dev/null
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/async/PollingAsyncItemProcessorMessagingGatewayTests-context.xml
@@ -0,0 +1,28 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+                                 http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+                                 http://www.springframework.org/schema/integration
+                                 http://www.springframework.org/schema/integration/spring-integration-2.0.xsd">
+	<annotation-config />
+	<channel id="requests">
+		<queue />
+		<interceptors>
+			<beans:bean class="org.springframework.batch.integration.async.StepExecutionInterceptor" />
+		</interceptors>
+	</channel>
+	<channel id="replies">
+		<queue />
+	</channel>
+	<gateway service-interface="org.springframework.batch.item.ItemProcessor" id="processor" default-reply-timeout="1000"
+		default-request-channel="requests" default-reply-channel="replies" />
+	<service-activator input-channel="requests" output-channel="replies" ref="doubler">
+		<poller fixed-rate="100" />
+	</service-activator>
+	<beans:bean id="doubler"
+		class="org.springframework.batch.integration.async.PollingAsyncItemProcessorMessagingGatewayTests$Doubler" />
+	<beans:bean class="org.springframework.batch.core.scope.StepScope">
+		<beans:property name="proxyTargetClass" value="true" />
+	</beans:bean>
+</beans:beans>
\ No newline at end of file