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 f83358b1c..32a18451a 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 @@ -29,6 +29,7 @@ import java.util.concurrent.CopyOnWriteArrayList; import javax.sql.DataSource; import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemStreamException; import org.springframework.beans.factory.InitializingBean; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.jdbc.core.RowCallbackHandler; @@ -48,17 +49,16 @@ import org.springframework.util.ClassUtils; * requested data. The query is executed using paged requests of a size * specified in {@link #setPageSize(int)}. Additional pages are requested when * needed as {@link #read()} method is called, returning an object corresponding - * to current position. + * 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 + * have been removed or modified). *

* *

* The performance of the paging depends on the database specific features - * available to limit the number of returned rows. - *

- * - *

- * Setting a fairly large page size and using a commit interval that matches the - * page size should provide better performance. + * available to limit the number of returned rows. Setting a fairly large page + * size and using a commit interval that matches the page size should provide + * better performance. *

* *

@@ -74,6 +74,11 @@ import org.springframework.util.ClassUtils; */ public class JdbcPagingItemReader extends AbstractPagingItemReader implements InitializingBean { + /** + * + */ + private static final String START_AFTER_VALUE = "start.after"; + public static final int VALUE_NOT_SET = -1; private DataSource dataSource; @@ -126,7 +131,9 @@ public class JdbcPagingItemReader extends AbstractPagingItemReader impleme } /** - * The row mapper implementation to be used by this reader + * The row mapper implementation to be used by this reader. The row mapper + * is used to convert result set rows into objects, which are then returned + * by the reader. * * @param rowMapper a * {@link org.springframework.jdbc.core.simple.ParameterizedRowMapper} @@ -217,9 +224,28 @@ public class JdbcPagingItemReader extends AbstractPagingItemReader impleme } @Override - protected void doJumpToPage(int itemIndex) { + public void update(ExecutionContext executionContext) throws ItemStreamException { + super.update(executionContext); + if (isSaveState() && startAfterValue != null) { + executionContext.put(getExecutionContextUserSupport().getKey(START_AFTER_VALUE), startAfterValue); + } + } - if (getPage() > 0) { + @Override + public void open(ExecutionContext executionContext) { + super.open(executionContext); + if (isSaveState()) { + startAfterValue = executionContext.get(getExecutionContextUserSupport().getKey(START_AFTER_VALUE)); + } + } + + @Override + protected void doJumpToPage(int itemIndex) { + /* + * Normally this would be false (the startAfterValue is enough + * information to restart from. + */ + if (startAfterValue == null && getPage() > 0) { String jumpToItemSql; jumpToItemSql = queryProvider.generateJumpToItemQuery(itemIndex, getPageSize()); @@ -243,7 +269,6 @@ public class JdbcPagingItemReader extends AbstractPagingItemReader impleme } } - } private Map getParameterMap(Map values, Object sortKeyValue) { diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemCountingItemStreamItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemCountingItemStreamItemReader.java index 26621a966..1ee240581 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemCountingItemStreamItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemCountingItemStreamItemReader.java @@ -34,8 +34,7 @@ import org.springframework.util.Assert; * * @author Robert Kasanicky */ -public abstract class AbstractItemCountingItemStreamItemReader implements - ItemReader, ItemStream { +public abstract class AbstractItemCountingItemStreamItemReader implements ItemReader, ItemStream { private static final String READ_COUNT = "read.count"; @@ -78,8 +77,7 @@ public abstract class AbstractItemCountingItemStreamItemReader implements } } - public final T read() throws Exception, UnexpectedInputException, - ParseException { + public final T read() throws Exception, UnexpectedInputException, ParseException { if (currentItemCount >= maxItemCount) { return null; } @@ -99,8 +97,7 @@ public abstract class AbstractItemCountingItemStreamItemReader implements * * @see #setName(String) * - * @param count - * the value of the current item count + * @param count the value of the current item count */ public void setCurrentItemCount(int count) { this.currentItemCount = count; @@ -108,14 +105,14 @@ public abstract class AbstractItemCountingItemStreamItemReader implements /** * The maximum index of the items to be read. If the - * {@link ExecutionContext} contains a key [name].read.count.max - * (where [name] is the name of this component) the value from - * the {@link ExecutionContext} will be used in preference. + * {@link ExecutionContext} contains a key + * [name].read.count.max (where [name] is the name + * of this component) the value from the {@link ExecutionContext} will be + * used in preference. * * @see #setName(String) * - * @param count - * the value of the maximum item count + * @param count the value of the maximum item count */ public void setMaxItemCount(int count) { this.maxItemCount = count; @@ -125,35 +122,34 @@ public abstract class AbstractItemCountingItemStreamItemReader implements currentItemCount = 0; try { doClose(); - } catch (Exception e) { + } + catch (Exception e) { throw new ItemStreamException("Error while closing item reader", e); } } - public void open(ExecutionContext executionContext) - throws ItemStreamException { + public void open(ExecutionContext executionContext) throws ItemStreamException { try { doOpen(); - } catch (Exception e) { + } + catch (Exception e) { throw new ItemStreamException("Failed to initialize the reader", e); } if (executionContext.containsKey(ecSupport.getKey(READ_COUNT_MAX))) { - maxItemCount = executionContext.getInt(ecSupport - .getKey(READ_COUNT_MAX)); + maxItemCount = executionContext.getInt(ecSupport.getKey(READ_COUNT_MAX)); } if (executionContext.containsKey(ecSupport.getKey(READ_COUNT))) { - int itemCount = executionContext.getInt(ecSupport - .getKey(READ_COUNT)); + int itemCount = executionContext.getInt(ecSupport.getKey(READ_COUNT)); if (itemCount < maxItemCount) { try { jumpToItem(itemCount); - } catch (Exception e) { - throw new ItemStreamException( - "Could not move to stored position on restart", e); + } + catch (Exception e) { + throw new ItemStreamException("Could not move to stored position on restart", e); } } currentItemCount = itemCount; @@ -162,28 +158,27 @@ public abstract class AbstractItemCountingItemStreamItemReader implements } - public void update(ExecutionContext executionContext) - throws ItemStreamException { + public void update(ExecutionContext executionContext) throws ItemStreamException { if (saveState) { - Assert.notNull(executionContext, - "ExecutionContext must not be null"); - executionContext.putInt(ecSupport.getKey(READ_COUNT), - currentItemCount); + Assert.notNull(executionContext, "ExecutionContext must not be null"); + executionContext.putInt(ecSupport.getKey(READ_COUNT), currentItemCount); if (maxItemCount < Integer.MAX_VALUE) { - executionContext.putInt(ecSupport.getKey(READ_COUNT_MAX), - maxItemCount); + executionContext.putInt(ecSupport.getKey(READ_COUNT_MAX), maxItemCount); } } } + protected ExecutionContextUserSupport getExecutionContextUserSupport() { + return ecSupport; + } + /** * The name of the component which will be used as a stem for keys in the * {@link ExecutionContext}. Subclasses should provide a default value, e.g. * the short form of the class name. * - * @param name - * the name for the component + * @param name the name for the component */ public void setName(String name) { ecSupport.setName(name); @@ -193,12 +188,21 @@ public abstract class AbstractItemCountingItemStreamItemReader implements * Set the flag that determines whether to save internal data for * {@link ExecutionContext}. Only switch this to false if you don't want to * save any state from this stream, and you don't need it to be restartable. + * Always set it to false if the reader is being used in a concurrent + * environment. * - * @param saveState - * flag value (default true). + * @param saveState flag value (default true). */ public void setSaveState(boolean saveState) { this.saveState = saveState; } + /** + * The flag that determines whether to save internal state for restarts. + * @return true if the flag was set + */ + public boolean isSaveState() { + return saveState; + } + } diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/iosample/JdbcPagingRestartIntegrationTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/iosample/JdbcPagingRestartIntegrationTests.java index 679bb8129..11019d9bb 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/iosample/JdbcPagingRestartIntegrationTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/iosample/JdbcPagingRestartIntegrationTests.java @@ -68,6 +68,7 @@ public class JdbcPagingRestartIntegrationTests { ExecutionContext executionContext = new ExecutionContext(); int count = SimpleJdbcTestUtils.countRowsInTable(jdbcTemplate, "CUSTOMER")-2; executionContext.putInt("JdbcPagingItemReader.read.count", count); + executionContext.putInt("JdbcPagingItemReader.start.after", 2); ((ItemStream)reader).open(executionContext); CustomerCredit item = reader.read(); // System.err.println(item);