diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java
index 4d492dfa1..2a34d2840 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java
@@ -94,6 +94,8 @@ public class FlatFileItemReader extends ExecutionContextUserSupport implements I
 	private LineTokenizer tokenizer = new DelimitedLineTokenizer();
 
 	private FieldSetMapper fieldSetMapper;
+	
+	private boolean saveState = false;
 
 	/**
 	 * Encapsulates the state of the input source. If it is null then we are uninitialized.
@@ -203,9 +205,12 @@ public class FlatFileItemReader extends ExecutionContextUserSupport implements I
 		if (reader == null) {
 			throw new ItemStreamException("ItemStream not open or already closed.");
 		}
-		Assert.notNull(executionContext, "ExecutionContext must not be null");
-		executionContext.putLong(getKey(LINES_READ_COUNT), reader.getPosition());
-		executionContext.putLong(getKey(SKIPPED_STATISTICS_NAME), skippedLines.size());
+		
+		if(saveState){
+			Assert.notNull(executionContext, "ExecutionContext must not be null");
+			executionContext.putLong(getKey(LINES_READ_COUNT), reader.getPosition());
+			executionContext.putLong(getKey(SKIPPED_STATISTICS_NAME), skippedLines.size());
+		}
 	}
 
 	/**
@@ -357,5 +362,17 @@ public class FlatFileItemReader extends ExecutionContextUserSupport implements I
 		Assert.notNull(resource, "Input resource must not be null");
 		Assert.notNull(fieldSetMapper, "FieldSetMapper must not be null.");
 	}
+	
+	/**
+	 * Set the boolean indicating whether or not state should be saved
+	 * in the provided {@link ExecutionContext} during the {@link ItemStream}
+	 * call to update.  Setting this to false means that it will always start
+	 * at the beginning.
+	 * 
+	 * @param saveState
+	 */
+	public void setSaveState(boolean saveState) {
+		this.saveState = saveState;
+	}
 
 }
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java
index 279a4678d..89b7424b3 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java
@@ -80,6 +80,8 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
 	private LineAggregator lineAggregator = new DelimitedLineAggregator();
 
 	private FieldSetCreator fieldSetCreator;
+	
+	private boolean saveState = false;
 
 	public FlatFileItemWriter() {
 		setName(ClassUtils.getShortName(FlatFileItemWriter.class));
@@ -191,16 +193,21 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
 		if (state == null) {
 			throw new ItemStreamException("ItemStream not open or already closed.");
 		}
+		
+
 		Assert.notNull(executionContext, "ExecutionContext must not be null");
-
-		try {
-			executionContext.putLong(getKey(RESTART_DATA_NAME), state.position());
-		} catch (IOException e) {
-			throw new ItemStreamException("ItemStream does not return current position properly", e);
+		
+		if(saveState){
+		
+			try {
+				executionContext.putLong(getKey(RESTART_DATA_NAME), state.position());
+			} catch (IOException e) {
+				throw new ItemStreamException("ItemStream does not return current position properly", e);
+			}
+	
+			executionContext.putLong(getKey(WRITTEN_STATISTICS_NAME), state.linesWritten);
+			executionContext.putLong(getKey(RESTART_COUNT_STATISTICS_NAME), state.restartCount);
 		}
-
-		executionContext.putLong(getKey(WRITTEN_STATISTICS_NAME), state.linesWritten);
-		executionContext.putLong(getKey(RESTART_COUNT_STATISTICS_NAME), state.restartCount);
 	}
 
 	// Returns object representing state.
@@ -467,5 +474,17 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implements I
 	public void flush() throws FlushFailedException {
 		getOutputState().mark();
 	}
+	
+	/**
+	 * Set the boolean indicating whether or not state should be saved
+	 * in the provided {@link ExecutionContext} during the {@link ItemStream}
+	 * call to update.  Setting this to false means that it will always start
+	 * at the beginning.
+	 * 
+	 * @param saveState
+	 */
+	public void setSaveState(boolean saveState) {
+		this.saveState = saveState;
+	}
 
 }
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemReaderAdvancedTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemReaderAdvancedTests.java
index a346d2dff..b462f23fc 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemReaderAdvancedTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemReaderAdvancedTests.java
@@ -65,6 +65,7 @@ public class FlatFileItemReaderAdvancedTests extends TestCase {
 		reader.setResource(getInputResource(TEST_STRING));
 		reader.setLineTokenizer(tokenizer);
 		reader.setFieldSetMapper(fieldSetMapper);
+		reader.setSaveState(true);
 		// context argument is necessary only for the FileLocator, which
 		// is mocked
 		executionContext = new ExecutionContext();
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemWriterTests.java
index 77dd90eb3..750613f62 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemWriterTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemWriterTests.java
@@ -71,6 +71,7 @@ public class FlatFileItemWriterTests extends TestCase {
 		inputSource.setResource(new FileSystemResource(outputFile));
 		inputSource.setFieldSetCreator(new PassThroughFieldSetMapper());
 		inputSource.afterPropertiesSet();
+		inputSource.setSaveState(true);
 		executionContext = new ExecutionContext();
 	}
 
@@ -267,6 +268,7 @@ public class FlatFileItemWriterTests extends TestCase {
 		inputSource.setResource(new FileSystemResource(outputFile));
 		inputSource.setFieldSetCreator(new PassThroughFieldSetMapper());
 		inputSource.afterPropertiesSet();
+		inputSource.setSaveState(true);
 		inputSource.open(executionContext);
 		inputSource.update(executionContext);
 		assertNotNull(executionContext);