Added builder for JdbcCursorItemReader

This provides a builder for the JdbcCursorItemReader to simplify java
configuration.

Resolves BATCH-2555

spring-batch-core/src/test/java/org/springframework/batch/core/resource/JdbcCursorItemReaderPreparedStatementIntegrationTests.java

@@ -28,6 +28,7 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.database.JdbcCursorItemReader;
+import org.springframework.batch.item.database.support.ListPreparedStatementSetter;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

spring-batch-core/src/test/java/org/springframework/batch/core/resource/ListPreparedStatementSetterTests.java

@@ -34,6 +34,7 @@ import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.job.AbstractJob;
 import org.springframework.batch.core.launch.JobLauncher;
 import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.item.database.support.ListPreparedStatementSetter;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.jdbc.core.RowCallbackHandler;

spring-batch-core/src/test/resources/org/springframework/batch/core/resource/ListPreparedStatementSetterTests-context.xml

@@ -30,7 +30,7 @@
 		</property>
 	</bean>
 
-	<bean id="pss" class="org.springframework.batch.core.resource.ListPreparedStatementSetter" scope="step">
+	<bean id="pss" class="org.springframework.batch.item.database.support.ListPreparedStatementSetter" scope="step">
 		<property name="parameters">
 			<list>
 				<value>#{jobParameters['min.id']}</value>

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/builder/JdbcCursorItemReaderBuilder.java

@@ -0,0 +1,353 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.item.database.builder;
+
+import java.util.List;
+import javax.sql.DataSource;
+
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.database.AbstractCursorItemReader;
+import org.springframework.batch.item.database.JdbcCursorItemReader;
+import org.springframework.batch.item.database.support.ListPreparedStatementSetter;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.jdbc.core.ArgumentPreparedStatementSetter;
+import org.springframework.jdbc.core.ArgumentTypePreparedStatementSetter;
+import org.springframework.jdbc.core.PreparedStatementSetter;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * Builder for the {@link JdbcCursorItemReader}
+ *
+ * @author Michael Minella
+ * @since 4.0
+ */
+public class JdbcCursorItemReaderBuilder<T> {
+
+	private DataSource dataSource;
+
+	private int fetchSize = AbstractCursorItemReader.VALUE_NOT_SET;
+
+	private int maxRows = AbstractCursorItemReader.VALUE_NOT_SET;
+
+	private int queryTimeout = AbstractCursorItemReader.VALUE_NOT_SET;
+
+	private int currentItemCount = 0;
+
+	private int maxItemCount = Integer.MAX_VALUE;
+
+	private boolean ignoreWarnings;
+
+	private boolean verifyCursorPosition;
+
+	private boolean driverSupportsAbsolute;
+
+	private boolean useSharedExtendedConnection;
+
+	private boolean saveState = true;
+
+	private PreparedStatementSetter preparedStatementSetter;
+
+	private String sql;
+
+	private String name;
+
+	private RowMapper<T> rowMapper;
+
+	/**
+	 * The {@link DataSource} to read from
+	 *
+	 * @param dataSource a relational data base
+	 * @return this instance for method chaining
+	 * @see JdbcCursorItemReader#setDataSource(DataSource)
+	 */
+	public JdbcCursorItemReaderBuilder<T> dataSource(DataSource dataSource) {
+		this.dataSource = dataSource;
+
+		return this;
+	}
+
+	/**
+	 * A hint to the driver as to how many rows to return with each fetch.
+	 *
+	 * @param fetchSize the hint
+	 * @return this instance for method chaining
+	 * @see JdbcCursorItemReader#setFetchSize(int)
+	 */
+	public JdbcCursorItemReaderBuilder<T> fetchSize(int fetchSize) {
+		this.fetchSize = fetchSize;
+
+		return this;
+	}
+
+	/**
+	 * The max number of rows the {@link java.sql.ResultSet} can contain
+	 *
+	 * @param maxRows the max
+	 * @return this instance for method chaining
+	 * @see JdbcCursorItemReader#setMaxRows(int)
+	 */
+	public JdbcCursorItemReaderBuilder<T> maxRows(int maxRows) {
+		this.maxRows = maxRows;
+
+		return this;
+	}
+
+	/**
+	 * The time in milliseconds for the query to timeout
+	 *
+	 * @param queryTimeout timeout
+	 * @return this instance for method chaining
+	 * @see JdbcCursorItemReader#setQueryTimeout(int)
+	 */
+	public JdbcCursorItemReaderBuilder<T> queryTimeout(int queryTimeout) {
+		this.queryTimeout = queryTimeout;
+
+		return this;
+	}
+
+	/**
+	 * The index of the first record to begin reading from.  Overridden if a previous value
+	 * is provided via the {@link org.springframework.batch.item.ExecutionContext} on
+	 * {@link org.springframework.batch.item.ItemStream#open(ExecutionContext)}
+	 *
+	 * @param currentItemCount current index
+	 * @return this instance for method chaining
+	 * @see JdbcCursorItemReader#setCurrentItemCount(int)
+	 */
+	public JdbcCursorItemReaderBuilder<T> currentItemCount(int currentItemCount) {
+		this.currentItemCount = currentItemCount;
+
+		return this;
+	}
+
+	/**
+	 * The max number of items to be read.  Overriden if a previous value is povided via
+	 * the {@link ExecutionContext} on {@link org.springframework.batch.item.ItemStream#open}
+	 *
+	 * @param maxItemCount count
+	 * @return this instance for method chaining
+	 * @see JdbcCursorItemReader#setMaxItemCount(int)
+	 */
+	public JdbcCursorItemReaderBuilder<T> maxItemCount(int maxItemCount) {
+		this.maxItemCount = maxItemCount;
+
+		return this;
+	}
+
+	/**
+	 * Indicates if the state of the reader should be persisted in the
+	 * {@link ExecutionContext}.  Defaults to true.
+	 *
+	 * @param saveState indicator.  Defaults to true
+	 * @return this instance for method chaining
+	 * @see JdbcCursorItemReader#setSaveState(boolean)
+	 */
+	public JdbcCursorItemReaderBuilder<T> saveState(boolean saveState) {
+		this.saveState = saveState;
+
+		return this;
+	}
+
+	public JdbcCursorItemReaderBuilder<T> ignoreWarnings(boolean ignoreWarnings) {
+		this.ignoreWarnings = ignoreWarnings;
+
+		return this;
+	}
+
+	/**
+	 * Indicates if the reader should verify the current position of the
+	 * {@link java.sql.ResultSet} after being passed to the {@link RowMapper}.  Defaults
+	 * to true.
+	 *
+	 * @param verifyCursorPosition indicator
+	 * @return this instance for method chaining
+	 * @see JdbcCursorItemReader#setVerifyCursorPosition(boolean)
+	 */
+	public JdbcCursorItemReaderBuilder<T> verifyCursorPosition(boolean verifyCursorPosition) {
+		this.verifyCursorPosition = verifyCursorPosition;
+
+		return this;
+	}
+
+	/**
+	 * Indicates if the JDBC driver supports setting the absolute row on the
+	 * {@link java.sql.ResultSet}.
+	 *
+	 * @param driverSupportsAbsolute indicator
+	 * @return this instance for method chaining
+	 * @see JdbcCursorItemReader#setDriverSupportsAbsolute(boolean)
+	 */
+	public JdbcCursorItemReaderBuilder<T> driverSupportsAbsolute(boolean driverSupportsAbsolute) {
+		this.driverSupportsAbsolute = driverSupportsAbsolute;
+
+		return this;
+	}
+
+	/**
+	 * Indicates that the connection used for the cursor is being used by all other
+	 * processing, therefor part of the same transaction.
+	 *
+	 * @param useSharedExtendedConnection indicator
+	 * @return this instance for method chaining
+	 * @see JdbcCursorItemReader#setUseSharedExtendedConnection(boolean)
+	 */
+	public JdbcCursorItemReaderBuilder<T> useSharedExtendedConnection(boolean useSharedExtendedConnection) {
+		this.useSharedExtendedConnection = useSharedExtendedConnection;
+
+		return this;
+	}
+
+	/**
+	 * Configures the provided {@link PreparedStatementSetter} to be used to populate any
+	 * arguments in the SQL query to be executed for the reader.
+	 *
+	 * @param preparedStatementSetter setter
+	 * @return this instance for method chaining
+	 * @see JdbcCursorItemReader#setPreparedStatementSetter(PreparedStatementSetter)
+	 */
+	public JdbcCursorItemReaderBuilder<T> preparedStatementSetter(PreparedStatementSetter preparedStatementSetter) {
+		this.preparedStatementSetter = preparedStatementSetter;
+
+		return this;
+	}
+
+	/**
+	 * Configures a {@link PreparedStatementSetter} that will use the array as the values
+	 * to be set on the query to be executed for this reader.
+	 *
+	 * @param args values to set on the reader query
+	 * @return this instance for method chaining
+	 */
+	public JdbcCursorItemReaderBuilder<T> queryArguments(Object[] args) {
+		this.preparedStatementSetter = new ArgumentPreparedStatementSetter(args);
+
+		return this;
+	}
+
+	/**
+	 * Configures a {@link PreparedStatementSetter} that will use the Object [] as the
+	 * values to be set on the query to be executed for this reader.  The int[] will
+	 * provide the types ({@link java.sql.Types}) for each of the values provided.
+	 *
+	 * @param args values to set on the query
+	 * @param types the type for each value in the args array
+	 * @return this instance for method chaining
+	 */
+	public JdbcCursorItemReaderBuilder<T> queryArguments(Object[] args, int[] types) {
+		this.preparedStatementSetter = new ArgumentTypePreparedStatementSetter(args, types);
+
+		return this;
+	}
+
+	/**
+	 * Configures a {@link PreparedStatementSetter} that will use the List as the values
+	 * to be set on the query to be executed for this reader.
+	 *
+	 * @param args values to set on the query
+	 * @return this instance for method chaining
+	 * @throws Exception from {@link InitializingBean#afterPropertiesSet()}
+	 */
+	public JdbcCursorItemReaderBuilder<T> queryArguments(List<?> args) throws Exception {
+		ListPreparedStatementSetter listPreparedStatementSetter = new ListPreparedStatementSetter(args);
+
+		listPreparedStatementSetter.afterPropertiesSet();
+
+		this.preparedStatementSetter = listPreparedStatementSetter;
+
+		return this;
+	}
+
+	/**
+	 * The query to be executed for this reader
+	 *
+	 * @param sql query
+	 * @return this instance for method chaining
+	 * @see JdbcCursorItemReader#setSql(String)
+	 */
+	public JdbcCursorItemReaderBuilder<T> sql(String sql) {
+		this.sql = sql;
+
+		return this;
+	}
+
+	/**
+	 * The {@link RowMapper} used to map the results of the cursor to each item.
+	 *
+	 * @param rowMapper {@link RowMapper}
+	 * @return this instance for method chaining
+	 * @see JdbcCursorItemReader#setRowMapper(RowMapper)
+	 */
+	public JdbcCursorItemReaderBuilder<T> rowMapper(RowMapper<T> rowMapper) {
+		this.rowMapper = rowMapper;
+
+		return this;
+	}
+
+	/**
+	 * A name used to prevent key collisions while saving state in the
+	 * {@link ExecutionContext}.
+	 *
+	 * @param name unique name for this reader instance
+	 * @return this instance for method chaining
+	 * @see JdbcCursorItemReader#setName(String)
+	 */
+	public JdbcCursorItemReaderBuilder<T> name(String name) {
+		this.name = name;
+
+		return this;
+	}
+
+	/**
+	 * Validates configuration and builds a new reader instance.
+	 *
+	 * @return a fully constructed {@link JdbcCursorItemReader}
+	 */
+	public JdbcCursorItemReader<T> build() {
+		if(this.saveState) {
+			Assert.hasText(this.name,
+					"A name is required when saveSate is set to true");
+		}
+
+		Assert.hasText(this.sql, "A query is required");
+		Assert.notNull(this.dataSource, "A datasource is required");
+		Assert.notNull(this.rowMapper, "A rowmapper is required");
+
+		JdbcCursorItemReader<T> reader = new JdbcCursorItemReader<>();
+
+		if(StringUtils.hasText(this.name)) {
+			reader.setName(this.name);
+		}
+
+		reader.setSaveState(this.saveState);
+		reader.setPreparedStatementSetter(this.preparedStatementSetter);
+		reader.setRowMapper(this.rowMapper);
+		reader.setSql(this.sql);
+		reader.setCurrentItemCount(this.currentItemCount);
+		reader.setDataSource(this.dataSource);
+		reader.setDriverSupportsAbsolute(this.driverSupportsAbsolute);
+		reader.setFetchSize(this.fetchSize);
+		reader.setIgnoreWarnings(this.ignoreWarnings);
+		reader.setMaxItemCount(this.maxItemCount);
+		reader.setMaxRows(this.maxRows);
+		reader.setQueryTimeout(this.queryTimeout);
+		reader.setUseSharedExtendedConnection(this.useSharedExtendedConnection);
+		reader.setVerifyCursorPosition(this.verifyCursorPosition);
+
+		return reader;
+	}
+ }

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/ListPreparedStatementSetter.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.batch.core.resource;
+package org.springframework.batch.item.database.support;
 
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
@@ -43,6 +43,12 @@ PreparedStatementSetter, InitializingBean {
 
 	private List<?> parameters;
 
+	public ListPreparedStatementSetter() {}
+
+	public ListPreparedStatementSetter(List<?> parameters) {
+		this.parameters = parameters;
+	}
+
 	@Override
 	public void setValues(PreparedStatement ps) throws SQLException {
 		for (int i = 0; i < parameters.size(); i++) {
@@ -54,7 +60,10 @@ PreparedStatementSetter, InitializingBean {
 	 * The parameter values that will be set on the PreparedStatement.
 	 * It is assumed that their order in the List is the order of the parameters
 	 * in the PreparedStatement.
+	 *
+	 * @deprecated In favor of the constructor
 	 */
+	@Deprecated
 	public void setParameters(List<?> parameters) {
 		this.parameters = parameters;
 	}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/builder/JdbcCursorItemReaderBuilderTests.java

@@ -0,0 +1,442 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.item.database.builder;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.Arrays;
+import javax.sql.DataSource;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.database.JdbcCursorItemReader;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.io.ByteArrayResource;
+import org.springframework.core.io.Resource;
+import org.springframework.jdbc.core.PreparedStatementSetter;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseFactory;
+import org.springframework.jdbc.datasource.init.DataSourceInitializer;
+import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * @author Michael Minella
+ */
+public class JdbcCursorItemReaderBuilderTests {
+
+	private DataSource dataSource;
+
+	private ConfigurableApplicationContext context;
+
+	@Before
+	public void setUp() {
+		this.context = new AnnotationConfigApplicationContext(TestDataSourceConfiguration.class);
+		this.dataSource = (DataSource) context.getBean("dataSource");
+	}
+
+	@After
+	public void tearDown() {
+		if(this.context != null) {
+			this.context.close();
+		}
+	}
+
+	@Test
+	public void testSimpleScenario() throws Exception {
+		JdbcCursorItemReader<Foo> reader = new JdbcCursorItemReaderBuilder<Foo>()
+				.dataSource(this.dataSource)
+				.name("fooReader")
+				.sql("SELECT * FROM FOO ORDER BY FIRST")
+				.rowMapper((rs, rowNum) -> {
+					Foo foo = new Foo();
+
+					foo.setFirst(rs.getInt("FIRST"));
+					foo.setSecond(rs.getString("SECOND"));
+					foo.setThird(rs.getString("THIRD"));
+
+					return foo;
+				})
+				.build();
+
+		ExecutionContext executionContext = new ExecutionContext();
+		reader.open(executionContext);
+
+		validateFoo(reader.read(), 1, "2", "3");
+		validateFoo(reader.read(), 4, "5", "6");
+		validateFoo(reader.read(), 7, "8", "9");
+
+		assertNull(reader.read());
+	}
+
+	@Test
+	public void testMaxRows() throws Exception {
+		JdbcCursorItemReader<Foo> reader = new JdbcCursorItemReaderBuilder<Foo>()
+				.dataSource(this.dataSource)
+				.name("fooReader")
+				.sql("SELECT * FROM FOO ORDER BY FIRST")
+				.maxRows(2)
+				.saveState(false)
+				.rowMapper((rs, rowNum) -> {
+					Foo foo = new Foo();
+
+					foo.setFirst(rs.getInt("FIRST"));
+					foo.setSecond(rs.getString("SECOND"));
+					foo.setThird(rs.getString("THIRD"));
+
+					return foo;
+				})
+				.build();
+
+		ExecutionContext executionContext = new ExecutionContext();
+		reader.open(executionContext);
+
+		validateFoo(reader.read(), 1, "2", "3");
+		validateFoo(reader.read(), 4, "5", "6");
+		assertNull(reader.read());
+
+		reader.close();
+		assertEquals(0, executionContext.size());
+	}
+
+	@Test
+	public void testQueryArgumentsList() throws Exception {
+		JdbcCursorItemReader<Foo> reader = new JdbcCursorItemReaderBuilder<Foo>()
+				.dataSource(this.dataSource)
+				.name("fooReader")
+				.sql("SELECT * FROM FOO WHERE FIRST > ? ORDER BY FIRST")
+				.queryArguments(Arrays.asList(3))
+				.rowMapper((rs, rowNum) -> {
+					Foo foo = new Foo();
+
+					foo.setFirst(rs.getInt("FIRST"));
+					foo.setSecond(rs.getString("SECOND"));
+					foo.setThird(rs.getString("THIRD"));
+
+					return foo;
+				})
+				.build();
+
+		ExecutionContext executionContext = new ExecutionContext();
+		reader.open(executionContext);
+
+		validateFoo(reader.read(), 4, "5", "6");
+		validateFoo(reader.read(), 7, "8", "9");
+
+		assertNull(reader.read());
+	}
+
+	@Test
+	public void testQueryArgumentsArray() throws Exception {
+		JdbcCursorItemReader<Foo> reader = new JdbcCursorItemReaderBuilder<Foo>()
+				.dataSource(this.dataSource)
+				.name("fooReader")
+				.sql("SELECT * FROM FOO WHERE FIRST > ? ORDER BY FIRST")
+				.queryArguments(new Integer[] {3})
+				.rowMapper((rs, rowNum) -> {
+					Foo foo = new Foo();
+
+					foo.setFirst(rs.getInt("FIRST"));
+					foo.setSecond(rs.getString("SECOND"));
+					foo.setThird(rs.getString("THIRD"));
+
+					return foo;
+				})
+				.build();
+
+		ExecutionContext executionContext = new ExecutionContext();
+		reader.open(executionContext);
+
+		validateFoo(reader.read(), 4, "5", "6");
+		validateFoo(reader.read(), 7, "8", "9");
+
+		assertNull(reader.read());
+	}
+
+	@Test
+	public void testQueryArgumentsTypedArray() throws Exception {
+		JdbcCursorItemReader<Foo> reader = new JdbcCursorItemReaderBuilder<Foo>()
+				.dataSource(this.dataSource)
+				.name("fooReader")
+				.sql("SELECT * FROM FOO WHERE FIRST > ? ORDER BY FIRST")
+				.queryArguments(new Integer[] {3}, new int[] {Types.BIGINT})
+				.rowMapper((rs, rowNum) -> {
+					Foo foo = new Foo();
+
+					foo.setFirst(rs.getInt("FIRST"));
+					foo.setSecond(rs.getString("SECOND"));
+					foo.setThird(rs.getString("THIRD"));
+
+					return foo;
+				})
+				.build();
+
+		ExecutionContext executionContext = new ExecutionContext();
+		reader.open(executionContext);
+
+		validateFoo(reader.read(), 4, "5", "6");
+		validateFoo(reader.read(), 7, "8", "9");
+
+		assertNull(reader.read());
+	}
+
+	@Test
+	public void testPreparedStatementSetter() throws Exception {
+		JdbcCursorItemReader<Foo> reader = new JdbcCursorItemReaderBuilder<Foo>()
+				.dataSource(this.dataSource)
+				.name("fooReader")
+				.sql("SELECT * FROM FOO WHERE FIRST > ? ORDER BY FIRST")
+				.preparedStatementSetter(new PreparedStatementSetter() {
+					@Override
+					public void setValues(PreparedStatement ps) throws SQLException {
+						ps.setInt(1, 3);
+					}
+				})
+				.rowMapper((rs, rowNum) -> {
+					Foo foo = new Foo();
+
+					foo.setFirst(rs.getInt("FIRST"));
+					foo.setSecond(rs.getString("SECOND"));
+					foo.setThird(rs.getString("THIRD"));
+
+					return foo;
+				})
+				.build();
+
+		ExecutionContext executionContext = new ExecutionContext();
+		reader.open(executionContext);
+
+		validateFoo(reader.read(), 4, "5", "6");
+		validateFoo(reader.read(), 7, "8", "9");
+
+		assertNull(reader.read());
+	}
+
+	@Test
+	public void testMaxItemCount() throws Exception {
+		JdbcCursorItemReader<Foo> reader = new JdbcCursorItemReaderBuilder<Foo>()
+				.dataSource(this.dataSource)
+				.name("fooReader")
+				.sql("SELECT * FROM FOO ORDER BY FIRST")
+				.maxItemCount(2)
+				.rowMapper((rs, rowNum) -> {
+					Foo foo = new Foo();
+
+					foo.setFirst(rs.getInt("FIRST"));
+					foo.setSecond(rs.getString("SECOND"));
+					foo.setThird(rs.getString("THIRD"));
+
+					return foo;
+				})
+				.build();
+
+		ExecutionContext executionContext = new ExecutionContext();
+		reader.open(executionContext);
+
+		validateFoo(reader.read(), 1, "2", "3");
+		validateFoo(reader.read(), 4, "5", "6");
+
+		assertNull(reader.read());
+	}
+
+	@Test
+	public void testCurrentItemCount() throws Exception {
+		JdbcCursorItemReader<Foo> reader = new JdbcCursorItemReaderBuilder<Foo>()
+				.dataSource(this.dataSource)
+				.name("fooReader")
+				.sql("SELECT * FROM FOO ORDER BY FIRST")
+				.currentItemCount(1)
+				.rowMapper((rs, rowNum) -> {
+					Foo foo = new Foo();
+
+					foo.setFirst(rs.getInt("FIRST"));
+					foo.setSecond(rs.getString("SECOND"));
+					foo.setThird(rs.getString("THIRD"));
+
+					return foo;
+				})
+				.build();
+
+		ExecutionContext executionContext = new ExecutionContext();
+		reader.open(executionContext);
+
+		validateFoo(reader.read(), 4, "5", "6");
+		validateFoo(reader.read(), 7, "8", "9");
+
+		assertNull(reader.read());
+	}
+
+	@Test
+	public void testOtherProperties() {
+		JdbcCursorItemReader<Foo> reader = new JdbcCursorItemReaderBuilder<Foo>()
+				.dataSource(this.dataSource)
+				.name("fooReader")
+				.sql("SELECT * FROM FOO ORDER BY FIRST")
+				.fetchSize(1)
+				.queryTimeout(2)
+				.ignoreWarnings(true)
+				.driverSupportsAbsolute(true)
+				.useSharedExtendedConnection(true)
+				.rowMapper((rs, rowNum) -> {
+					Foo foo = new Foo();
+
+					foo.setFirst(rs.getInt("FIRST"));
+					foo.setSecond(rs.getString("SECOND"));
+					foo.setThird(rs.getString("THIRD"));
+
+					return foo;
+				})
+				.build();
+
+		assertEquals(1, ReflectionTestUtils.getField(reader, "fetchSize"));
+		assertEquals(2, ReflectionTestUtils.getField(reader, "queryTimeout"));
+		assertTrue((boolean) ReflectionTestUtils.getField(reader, "ignoreWarnings"));
+		assertTrue((boolean) ReflectionTestUtils.getField(reader, "driverSupportsAbsolute"));
+	}
+
+	@Test
+	public void testValidation() {
+		try {
+			new JdbcCursorItemReaderBuilder<Foo>().saveState(true).build();
+		}
+		catch (IllegalArgumentException iae) {
+			assertEquals("A name is required when saveSate is set to true", iae.getMessage());
+		}
+		catch (Exception e) {
+			fail();
+		}
+
+		try {
+			new JdbcCursorItemReaderBuilder<Foo>()
+					.saveState(false)
+					.build();
+		}
+		catch (IllegalArgumentException iae) {
+			assertEquals("A query is required", iae.getMessage());
+		}
+		catch (Exception e) {
+			fail();
+		}
+
+		try {
+			new JdbcCursorItemReaderBuilder<Foo>()
+					.saveState(false)
+					.sql("select 1")
+					.build();
+		}
+		catch (IllegalArgumentException iae) {
+			assertEquals("A datasource is required", iae.getMessage());
+		}
+		catch (Exception e) {
+			fail();
+		}
+
+		try {
+			new JdbcCursorItemReaderBuilder<Foo>()
+					.saveState(false)
+					.sql("select 1")
+					.dataSource(this.dataSource)
+					.build();
+		}
+		catch (IllegalArgumentException iae) {
+			assertEquals("A rowmapper is required", iae.getMessage());
+		}
+		catch (Exception e) {
+			fail();
+		}
+	}
+
+	private void validateFoo(Foo item, int first, String second, String third) {
+		assertEquals(first, item.getFirst());
+		assertEquals(second, item.getSecond());
+		assertEquals(third, item.getThird());
+	}
+
+	public static class Foo {
+		private int first;
+		private String second;
+		private String third;
+
+		public int getFirst() {
+			return first;
+		}
+
+		public void setFirst(int first) {
+			this.first = first;
+		}
+
+		public String getSecond() {
+			return second;
+		}
+
+		public void setSecond(String second) {
+			this.second = second;
+		}
+
+		public String getThird() {
+			return third;
+		}
+
+		public void setThird(String third) {
+			this.third = third;
+		}
+	}
+
+	@Configuration
+	public static class TestDataSourceConfiguration {
+
+		private static final String CREATE_SQL = "CREATE TABLE FOO  (\n" +
+				"\tID BIGINT IDENTITY NOT NULL PRIMARY KEY ,\n" +
+				"\tFIRST BIGINT ,\n" +
+				"\tSECOND VARCHAR(5) NOT NULL,\n" +
+				"\tTHIRD VARCHAR(5) NOT NULL) ;";
+
+		private static final String INSERT_SQL =
+				"INSERT INTO FOO (FIRST, SECOND, THIRD) VALUES (1, '2', '3');" +
+				"INSERT INTO FOO (FIRST, SECOND, THIRD) VALUES (4, '5', '6');" +
+				"INSERT INTO FOO (FIRST, SECOND, THIRD) VALUES (7, '8', '9');";
+
+		@Bean
+		public DataSource dataSource() {
+			return new EmbeddedDatabaseFactory().getDatabase();
+		}
+
+		@Bean
+		public DataSourceInitializer initializer(DataSource dataSource) {
+			DataSourceInitializer dataSourceInitializer = new DataSourceInitializer();
+			dataSourceInitializer.setDataSource(dataSource);
+
+			Resource create = new ByteArrayResource(CREATE_SQL.getBytes());
+			Resource insert = new ByteArrayResource(INSERT_SQL.getBytes());
+			dataSourceInitializer.setDatabasePopulator(new ResourceDatabasePopulator(create, insert));
+
+			return dataSourceInitializer;
+		}
+
+	}
+
+}