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 eb871790e..d13dff26c 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
@@ -23,7 +23,6 @@ import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemStreamException;
 import org.springframework.batch.item.ItemStreamSupport;
 import org.springframework.batch.item.support.CompositeItemStream;
-import org.springframework.util.ClassUtils;
 
 /**
  * Manage the offset data between the last successful commit and updates made to
@@ -59,10 +58,10 @@ public class ChunkMonitor extends ItemStreamSupport {
 
 	private ItemReader<?> reader;
 
-        public ChunkMonitor() {
-                this.setExecutionContextName(ChunkMonitor.class.getName());
-        }
-        
+	public ChunkMonitor() {
+		this.setExecutionContextName(ChunkMonitor.class.getName());
+	}
+
 	/**
 	 * @param stream the stream to set
 	 */
@@ -101,7 +100,7 @@ public class ChunkMonitor extends ItemStreamSupport {
 
 	@Override
 	public void close() throws ItemStreamException {
-                super.close();
+		super.close();
 		holder.set(null);
 		if (streamsRegistered) {
 			stream.close();
@@ -110,7 +109,7 @@ public class ChunkMonitor extends ItemStreamSupport {
 
 	@Override
 	public void open(ExecutionContext executionContext) throws ItemStreamException {
-                super.open(executionContext);
+		super.open(executionContext);
 		if (streamsRegistered) {
 			stream.open(executionContext);
 			ChunkMonitorData data = new ChunkMonitorData(executionContext.getInt(getExecutionContextKey(OFFSET), 0), 0);
@@ -127,12 +126,14 @@ public class ChunkMonitor extends ItemStreamSupport {
 					throw new ItemStreamException("Could not position reader with offset: " + data.offset, e);
 				}
 			}
+
+			resetOffset();
 		}
 	}
 
 	@Override
 	public void update(ExecutionContext executionContext) throws ItemStreamException {
-                super.update(executionContext);
+		super.update(executionContext);
 		if (streamsRegistered) {
 			ChunkMonitorData data = getData();
 			if (data.offset == 0) {
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
index c3956ac49..55d9e4d8b 100755
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
@@ -256,7 +256,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 					}
 					if (output == null) {
 						// No need to re-process filtered items
-						iterator.remove();						
+						iterator.remove();
 					}
 					return output;
 				}
@@ -313,7 +313,6 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 	@Override
 	protected void write(final StepContribution contribution, final Chunk<I> inputs, final Chunk<O> outputs)
 			throws Exception {
-
 		@SuppressWarnings("unchecked")
 		final UserData<O> data = (UserData<O>) inputs.getUserData();
 		final AtomicReference<RetryContext> contextHolder = new AtomicReference<RetryContext>();
@@ -568,6 +567,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 		if (outputs.isEmpty()) {
 			data.scanning(false);
 			inputs.setBusy(false);
+			chunkMonitor.resetOffset();
 			return;
 		}
 
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 4b8edf3cc..3d8c6be30 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
@@ -55,7 +55,7 @@ public class ChunkMonitorTests {
 		monitor.registerItemStream(new ItemStreamSupport() {
 			@Override
 			public void close() {
-                                super.close();
+				super.close();
 				closed = true;
 			}
 		});
@@ -98,6 +98,7 @@ public class ChunkMonitorTests {
 		executionContext.putInt(ChunkMonitor.class.getName() + ".OFFSET", 2);
 		monitor.open(executionContext);
 		assertEquals(2, count);
+		assertEquals(0, monitor.getOffset());
 	}
 
 	@Test