diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java
index 782bf6460..d0e3db9d7 100755
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java
@@ -262,8 +262,9 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 	protected void registerStreams(TaskletStep step, ItemStream[] streams) {
 		CompositeItemStream composite = new CompositeItemStream();
 		boolean streamIsReader = false;
-		for (final ItemStream stream : streams) {
-			if (stream instanceof ItemReader) {
+		ItemReader<? extends T> itemReader = getItemReader();
+		for (ItemStream stream : streams) {
+			if (stream == itemReader) {
 				streamIsReader = true;
 				composite.register(stream);
 			}
@@ -279,7 +280,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 			step.registerStream(chunkMonitor);
 			boolean concurrent = taskExecutor != null && !(taskExecutor instanceof SyncTaskExecutor);
 			if (!concurrent) {
-				chunkMonitor.setItemReader(getItemReader());
+				chunkMonitor.setItemReader(itemReader);
 			}
 			else {
 				logger.warn("Synchronous TaskExecutor detected (" + taskExecutor.getClass()
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java
index c9fc27176..6d43e531c 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java
@@ -33,6 +33,7 @@ import org.springframework.batch.core.step.skip.SkipPolicy;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemStreamException;
 import org.springframework.batch.item.ItemStreamReader;
 import org.springframework.batch.item.ItemWriter;
@@ -695,6 +696,56 @@ public class FaultTolerantStepFactoryBeanTests {
 		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
 	}
 
+	/**
+	 * Check ItemStream is opened
+	 */
+	@Test
+	public void testNestedItemStreamOpened() throws Exception {
+
+		ItemStreamReader<String> reader = new ItemStreamReader<String>() {
+			public void close() throws ItemStreamException {
+			}
+
+			public void open(ExecutionContext executionContext) throws ItemStreamException {
+			}
+
+			public void update(ExecutionContext executionContext) throws ItemStreamException {
+			}
+
+			public String read() throws Exception, UnexpectedInputException, ParseException {
+				return null;
+			}
+		};
+
+		ItemStreamReader<String> stream = new ItemStreamReader<String>() {
+			public void close() throws ItemStreamException {
+				closed = true;
+			}
+
+			public void open(ExecutionContext executionContext) throws ItemStreamException {
+				opened = true;
+			}
+
+			public void update(ExecutionContext executionContext) throws ItemStreamException {
+			}
+
+			public String read() throws Exception, UnexpectedInputException, ParseException {
+				return null;
+			}
+		};
+
+		factory.setItemReader(reader);
+		factory.setStreams(new ItemStream[] {stream, reader});
+
+		Step step = (Step) factory.getObject();
+
+		step.execute(stepExecution);
+
+		assertTrue(opened);
+		assertTrue(closed);
+		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
+	}
+	
 	private static class SkipProcessorStub implements ItemProcessor<String, String> {
 		private final Collection<String> failures;