diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java
index 98a445450..8984415b4 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java
@@ -21,6 +21,7 @@ import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemStreamException;
+import org.springframework.batch.item.support.CompositeItemStream;
 
 /**
  * Manage the offset data between the last successful commit and updates made to
@@ -48,7 +49,7 @@ class ChunkMonitor implements ItemStream {
 
 	private static final String OFFSET = ChunkMonitor.class.getName() + ".OFFSET";
 
-	private ItemStream stream;
+	private CompositeItemStream stream = new CompositeItemStream();
 
 	private ThreadLocal<ChunkMonitorData> holder = new ThreadLocal<ChunkMonitorData>();
 	{
@@ -62,8 +63,8 @@ class ChunkMonitor implements ItemStream {
 	/**
 	 * @param stream the stream to set
 	 */
-	public void setItemStream(ItemStream stream) {
-		this.stream = stream;
+	public void registerItemStream(ItemStream stream) {
+		this.stream.register(stream);
 	}
 
 	/**
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 d0e3db9d7..eb932568d 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
@@ -33,7 +33,6 @@ import org.springframework.batch.core.step.skip.SkipPolicy;
 import org.springframework.batch.core.step.tasklet.TaskletStep;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemStream;
-import org.springframework.batch.item.support.CompositeItemStream;
 import org.springframework.batch.repeat.RepeatOperations;
 import org.springframework.batch.repeat.support.RepeatTemplate;
 import org.springframework.batch.retry.RetryException;
@@ -260,13 +259,12 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 	 */
 	@Override
 	protected void registerStreams(TaskletStep step, ItemStream[] streams) {
-		CompositeItemStream composite = new CompositeItemStream();
 		boolean streamIsReader = false;
 		ItemReader<? extends T> itemReader = getItemReader();
 		for (ItemStream stream : streams) {
-			if (stream == itemReader) {
+			if (stream instanceof ItemReader) {
 				streamIsReader = true;
-				composite.register(stream);
+				chunkMonitor.registerItemStream(stream);
 			}
 			else {
 				step.registerStream(stream);
@@ -276,7 +274,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 		// In cases where multiple nested item readers are registered,
 		// they all want to get the open() and close() callbacks.
 		if (streamIsReader) {
-			chunkMonitor.setItemStream(composite);
+			// double registration is fine
 			step.registerStream(chunkMonitor);
 			boolean concurrent = taskExecutor != null && !(taskExecutor instanceof SyncTaskExecutor);
 			if (!concurrent) {
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkMonitorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkMonitorTests.java
index 4296897f6..9f590d779 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkMonitorTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkMonitorTests.java
@@ -51,7 +51,7 @@ public class ChunkMonitorTests {
 				return "" + (count++);
 			}
 		});
-		monitor.setItemStream(new ItemStreamSupport() {
+		monitor.registerItemStream(new ItemStreamSupport() {
 			@Override
 			public void close() throws ItemStreamException {
 				closed = true;