From d78454361ce020bc294501b1b34620c34851ab61 Mon Sep 17 00:00:00 2001
From: jpraet
Date: Fri, 12 Jul 2013 21:06:02 +0200
Subject: [PATCH] BATCH-1849: Item was not picked up after restarting a failed
job!!!
---
.../item/database/JdbcPagingItemReader.java | 19 +++++++++--
...tDataSourceItemReaderIntegrationTests.java | 34 +++++++++++++++++++
2 files changed, 50 insertions(+), 3 deletions(-)
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).
*
*
@@ -94,6 +94,8 @@ public class JdbcPagingItemReader extends AbstractPagingItemReader impleme
private String remainingPagesSql;
private Map startAfterValues;
+
+ private Map previousStartAfterValues;
private int fetchSize = VALUE_NOT_SET;
@@ -210,6 +212,7 @@ public class JdbcPagingItemReader extends AbstractPagingItemReader impleme
}
else {
+ previousStartAfterValues = startAfterValues;
if (logger.isDebugEnabled()) {
logger.debug("SQL used for reading remaining pages: [" + remainingPagesSql + "]");
}
@@ -230,10 +233,20 @@ public class JdbcPagingItemReader extends AbstractPagingItemReader 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.
*/