diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java
index 9074fb1b5..3ba6c8089 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java
@@ -50,7 +50,7 @@ import org.springframework.util.ClassUtils;
  * specified in {@link #setPageSize(int)}. Additional pages are requested when
  * needed as {@link #read()} method is called, returning an object corresponding
  * to current position. On restart it uses the last sort key value to locate the
- * first page to read (so it doesn't matter if the successfully processed itmes
+ * first page to read (so it doesn't matter if the successfully processed items
  * have been removed or modified).
  * </p>
  * 
@@ -94,6 +94,8 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
 	private String remainingPagesSql;
 
 	private Map<String, Object> startAfterValues;
+	
+	private Map<String, Object> previousStartAfterValues;
 
 	private int fetchSize = VALUE_NOT_SET;
 
@@ -210,6 +212,7 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
 
 		}
 		else {
+			previousStartAfterValues = startAfterValues;
 			if (logger.isDebugEnabled()) {
 				logger.debug("SQL used for reading remaining pages: [" + remainingPagesSql + "]");
 			}
@@ -230,10 +233,20 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
 	@Override
 	public void update(ExecutionContext executionContext) throws ItemStreamException {
 		super.update(executionContext);
-		if (isSaveState() && startAfterValues != null) {
-			executionContext.put(getExecutionContextKey(START_AFTER_VALUE), startAfterValues);
+		if (isSaveState()) {
+			if (isAtEndOfPage() && startAfterValues != null) {
+				// restart on next page
+				executionContext.put(getExecutionContextKey(START_AFTER_VALUE), startAfterValues);	
+			} else if (previousStartAfterValues != null) {
+				// restart on current page
+				executionContext.put(getExecutionContextKey(START_AFTER_VALUE), previousStartAfterValues);
+			}
 		}
 	}
+	
+	private boolean isAtEndOfPage() {
+		return getCurrentItemCount() % getPageSize() == 0;
+	}
 
 	@Override
 	@SuppressWarnings("unchecked")
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/AbstractDataSourceItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/AbstractDataSourceItemReaderIntegrationTests.java
index e792e6778..94bbab6eb 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/AbstractDataSourceItemReaderIntegrationTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/AbstractDataSourceItemReaderIntegrationTests.java
@@ -111,6 +111,40 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests {
 		assertEquals(3, fooAfterRestart.getValue());
 	}
 
+	/*
+	 * Restart scenario - read records, save restart data, create new input
+	 * source and restore from restart data - the new input source should
+	 * continue where the old one finished.
+	 */
+	@Transactional @Test
+	public void testRestartOnSecondPage() throws Exception {
+
+		getAsItemStream(reader).open(executionContext);
+		
+		Foo foo1 = reader.read();
+		assertEquals(1, foo1.getValue());
+		Foo foo2 = reader.read();
+		assertEquals(2, foo2.getValue());
+		Foo foo3 = reader.read();
+		assertEquals(3, foo3.getValue());
+		Foo foo4 = reader.read();
+		assertEquals(4, foo4.getValue());
+
+		getAsItemStream(reader).update(executionContext);
+	
+		getAsItemStream(reader).close();
+
+		// create new input source
+		reader = createItemReader();
+
+		getAsItemStream(reader).open(executionContext);
+
+		Foo foo5 = reader.read();
+		assertEquals(5, foo5.getValue());
+
+		assertNull(reader.read());
+	}
+
 	/*
 	 * Reading from an input source and then trying to restore causes an error.
 	 */