diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractCursorItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractCursorItemReader.java
new file mode 100644
index 000000000..f07cd0ade
--- /dev/null
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractCursorItemReader.java
@@ -0,0 +1,485 @@
+/*
+ * Copyright 2006-2009 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;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.SQLWarning;
+import java.sql.Statement;
+
+import javax.sql.DataSource;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.ItemStream;
+import org.springframework.batch.item.ReaderNotOpenException;
+import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.dao.InvalidDataAccessResourceUsageException;
+import org.springframework.jdbc.SQLWarningException;
+import org.springframework.jdbc.datasource.DataSourceUtils;
+import org.springframework.jdbc.support.JdbcUtils;
+import org.springframework.jdbc.support.SQLErrorCodeSQLExceptionTranslator;
+import org.springframework.jdbc.support.SQLExceptionTranslator;
+import org.springframework.jdbc.support.SQLStateSQLExceptionTranslator;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.util.Assert;
+
+/**
+ * <p>
+ * Abstract base class for any simple item reader that opens a database cursor and continually retrieves 
+ * the next row in the ResultSet. 
+ * </p>
+ * 
+ * <p>
+ * By default the cursor will be opened using a separate connection. The ResultSet for the cursor
+ * is held open regardless of commits or roll backs in a surrounding transaction. Clients of this 
+ * reader are responsible for buffering the items in the case that they need to be re-presented on a
+ * rollback. This buffering is handled by the step implementations provided and is only a concern for
+ * anyone writing their own step implementations.
+ * </p>
+ * 
+ * <p>
+ * There is an option ({@link #setUseSharedExtendedConnection(boolean)} that will share the connection 
+ * used for the cursor with the rest of the step processing. If you set this flag to <code>true</code> 
+ * then you must wrap the DataSource in a {@link ExtendedConnectionDataSourceProxy} to prevent the 
+ * connection from being closed and released after each commit performed as part of the step processing. 
+ * You must also use a JDBC driver supporting JDBC 3.0 or later since the cursor will be opened with the 
+ * additional option of 'HOLD_CUSORS_OVER_COMMIT' enabled. 
+ * </p>
+ * 
+ * <p>
+ * Each call to {@link #read()} will attempt to map the row at the current position in the
+ * ResultSet. There is currently no wrapping of the ResultSet to suppress calls
+ * to next(). However, if the RowMapper (mistakenly) increments the current row,
+ * the next call to read will verify that the current row is at the expected
+ * position and throw a DataAccessException if it is not. The reason for such strictness on the
+ * ResultSet is due to the need to maintain control for transactions and
+ * restartability. This ensures that each call to {@link #read()} returns the
+ * ResultSet at the correct row, regardless of rollbacks or restarts.
+ * </p>
+ * 
+ * <p>
+ * {@link ExecutionContext}: The current row is returned as restart data, and
+ * when restored from that same data, the cursor is opened and the current row
+ * set to the value within the restart data. See
+ * {@link #setDriverSupportsAbsolute(boolean)} for improving restart
+ * performance.
+ * </p>
+ * 
+ * <p>
+ * Calling close on this {@link ItemStream} will cause all resources it is
+ * currently using to be freed. (Connection, ResultSet, etc). It is then illegal
+ * to call {@link #read()} again until it has been re-opened.
+ * </p>
+ * 
+ * <p>
+ * Known limitation: when used with Derby
+ * {@link #setVerifyCursorPosition(boolean)} needs to be <code>false</code>
+ * because {@link ResultSet#getRow()} call used for cursor position verification
+ * is not available for 'TYPE_FORWARD_ONLY' result sets.
+ * </p>
+ * 
+ * @author Lucas Ward
+ * @author Peter Zozom
+ * @author Robert Kasanicky
+ * @author Thomas Risberg
+ */
+public abstract class AbstractCursorItemReader<T> extends AbstractItemCountingItemStreamItemReader<T>
+		implements InitializingBean {
+
+	/** Logger available to subclasses */
+	protected final Log log = LogFactory.getLog(getClass());
+
+	public static final int VALUE_NOT_SET = -1;
+	private Connection con;
+	
+	protected ResultSet rs;
+	
+	private DataSource dataSource;
+	
+	private int fetchSize = VALUE_NOT_SET;
+	
+	private int maxRows = VALUE_NOT_SET;
+	
+	private int queryTimeout = VALUE_NOT_SET;
+	
+	private boolean ignoreWarnings = true;
+	
+	private boolean verifyCursorPosition = true;
+	
+	private SQLExceptionTranslator exceptionTranslator;
+	
+	private boolean initialized = false;
+	
+	private boolean driverSupportsAbsolute = false;
+	
+	private boolean useSharedExtendedConnection = false;
+	
+
+	public AbstractCursorItemReader() {
+		super();
+	}
+
+	/**
+	 * Assert that mandatory properties are set.
+	 * 
+	 * @throws IllegalArgumentException if either data source or sql properties
+	 * not set.
+	 */
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(dataSource, "DataSource must be provided");
+	}
+
+	/**
+	 * Public setter for the data source for injection purposes.
+	 * 
+	 * @param dataSource
+	 */
+	public void setDataSource(DataSource dataSource) {
+		this.dataSource = dataSource;
+	}
+
+	/**
+	 * Public getter for the data source.
+	 * 
+	 * @return the dataSource
+	 */
+	public DataSource getDataSource() {
+		return this.dataSource;
+	}
+
+	/**
+	 * Prepare the given JDBC Statement (or PreparedStatement or
+	 * CallableStatement), applying statement settings such as fetch size, max
+	 * rows, and query timeout. @param stmt the JDBC Statement to prepare
+	 * @throws SQLException
+	 * 
+	 * @see #setFetchSize
+	 * @see #setMaxRows
+	 * @see #setQueryTimeout
+	 */
+	protected void applyStatementSettings(PreparedStatement stmt) throws SQLException {
+		if (fetchSize != VALUE_NOT_SET) {
+			stmt.setFetchSize(fetchSize);
+			stmt.setFetchDirection(ResultSet.FETCH_FORWARD);
+		}
+		if (maxRows != VALUE_NOT_SET) {
+			stmt.setMaxRows(maxRows);
+		}
+		if (queryTimeout != VALUE_NOT_SET) {
+			stmt.setQueryTimeout(queryTimeout);
+		}
+	}
+
+	/**
+	 * Return the exception translator for this instance.
+	 * 
+	 * Creates a default SQLErrorCodeSQLExceptionTranslator for the specified
+	 * DataSource if none is set.
+	 */
+	protected SQLExceptionTranslator getExceptionTranslator() {
+		synchronized(this) {
+			if (exceptionTranslator == null) {
+				if (dataSource != null) {
+					exceptionTranslator = new SQLErrorCodeSQLExceptionTranslator(dataSource);
+				}
+				else {
+					exceptionTranslator = new SQLStateSQLExceptionTranslator();
+				}
+			}
+		}
+		return exceptionTranslator;
+	}
+
+	/**
+	 * Throw a SQLWarningException if we're not ignoring warnings, else log the
+	 * warnings (at debug level).
+	 * 
+	 * @param statement the current statement to obtain the warnings from, if there are any.
+	 * @throws SQLException
+	 * 
+	 * @see org.springframework.jdbc.SQLWarningException
+	 */
+	protected void handleWarnings(Statement statement) throws SQLWarningException,
+			SQLException {
+				if (ignoreWarnings) {
+					if (log.isDebugEnabled()) {
+						SQLWarning warningToLog = statement.getWarnings();
+						while (warningToLog != null) {
+							log.debug("SQLWarning ignored: SQL state '" + warningToLog.getSQLState() + "', error code '"
+									+ warningToLog.getErrorCode() + "', message [" + warningToLog.getMessage() + "]");
+							warningToLog = warningToLog.getNextWarning();
+						}
+					}
+				}
+				else {
+					SQLWarning warnings = statement.getWarnings();
+					if (warnings != null) {
+						throw new SQLWarningException("Warning not ignored", warnings);
+					}
+				}
+			}
+
+	/**
+	 * Moves the cursor in the ResultSet to the position specified by the row
+	 * parameter by traversing the ResultSet.
+	 * @param row
+	 */
+	private void moveCursorToRow(int row) {
+		try {
+			int count = 0;
+			while (row != count && rs.next()) {
+				count++;
+			}
+		}
+		catch (SQLException se) {
+			throw getExceptionTranslator().translate("Attempted to move ResultSet to last committed row", getSql(), se);
+		}
+	}
+
+	/**
+	 * Gives the JDBC driver a hint as to the number of rows that should be
+	 * fetched from the database when more rows are needed for this
+	 * <code>ResultSet</code> object. If the fetch size specified is zero, the
+	 * JDBC driver ignores the value.
+	 * 
+	 * @param fetchSize the number of rows to fetch
+	 * @see ResultSet#setFetchSize(int)
+	 */
+	public void setFetchSize(int fetchSize) {
+		this.fetchSize = fetchSize;
+	}
+
+	/**
+	 * Sets the limit for the maximum number of rows that any
+	 * <code>ResultSet</code> object can contain to the given number.
+	 * 
+	 * @param maxRows the new max rows limit; zero means there is no limit
+	 * @see Statement#setMaxRows(int)
+	 */
+	public void setMaxRows(int maxRows) {
+		this.maxRows = maxRows;
+	}
+
+	/**
+	 * Sets the number of seconds the driver will wait for a
+	 * <code>Statement</code> object to execute to the given number of seconds.
+	 * If the limit is exceeded, an <code>SQLException</code> is thrown.
+	 * 
+	 * @param queryTimeout seconds the new query timeout limit in seconds; zero
+	 * means there is no limit
+	 * @see Statement#setQueryTimeout(int)
+	 */
+	public void setQueryTimeout(int queryTimeout) {
+		this.queryTimeout = queryTimeout;
+	}
+
+	/**
+	 * Set whether SQLWarnings should be ignored (only logged) or exception
+	 * should be thrown.
+	 * 
+	 * @param ignoreWarnings if TRUE, warnings are ignored
+	 */
+	public void setIgnoreWarnings(boolean ignoreWarnings) {
+		this.ignoreWarnings = ignoreWarnings;
+	}
+
+	/**
+	 * Allow verification of cursor position after current row is processed by
+	 * RowMapper or RowCallbackHandler. Default value is TRUE.
+	 * 
+	 * @param verifyCursorPosition if true, cursor position is verified
+	 */
+	public void setVerifyCursorPosition(boolean verifyCursorPosition) {
+		this.verifyCursorPosition = verifyCursorPosition;
+	}
+
+	/**
+	 * Indicate whether the JDBC driver supports setting the absolute row on a
+	 * {@link ResultSet}. It is recommended that this is set to
+	 * <code>true</code> for JDBC drivers that supports ResultSet.absolute() as
+	 * it may improve performance, especially if a step fails while working with
+	 * a large data set.
+	 * 
+	 * @see ResultSet#absolute(int)
+	 * 
+	 * @param driverSupportsAbsolute <code>false</code> by default
+	 */
+	public void setDriverSupportsAbsolute(boolean driverSupportsAbsolute) {
+		this.driverSupportsAbsolute = driverSupportsAbsolute;
+	}
+
+	/**
+	 * Indicate whether the connection used for the cursor should be used by all other processing
+	 * thus sharing the same transaction. If this is set to false, which is the default, then the 
+	 * cursor will be opened using in its connection and will not participate in any transactions
+	 * started for the rest of the step processing. If you set this flag to true then you must 
+	 * wrap the DataSource in a {@link ExtendedConnectionDataSourceProxy} to prevent the
+	 * connection from being closed and released after each commit.
+	 * 
+	 * When you set this option to <code>true</code> then the statement used to open the cursor 
+	 * will be created with both 'READ_ONLY' and 'HOLD_CUSORS_OVER_COMMIT' options. This allows 
+	 * holding the cursor open over transaction start and commits performed in the step processing. 
+	 * To use this feature you need a database that supports this and a JDBC driver supporting 
+	 * JDBC 3.0 or later.
+	 *
+	 * @param useSharedExtendedConnection <code>false</code> by default
+	 */
+	public void setUseSharedExtendedConnection(boolean useSharedExtendedConnection) {
+		this.useSharedExtendedConnection = useSharedExtendedConnection;
+	}
+
+	public boolean isUseSharedExtendedConnection() {
+		return useSharedExtendedConnection;
+	}
+
+	public abstract String getSql();
+	
+	/**
+	 * Check the result set is in synch with the currentRow attribute. This is
+	 * important to ensure that the user hasn't modified the current row.
+	 */
+	private void verifyCursorPosition(long expectedCurrentRow) throws SQLException {
+		if (verifyCursorPosition) {
+			if (expectedCurrentRow != this.rs.getRow()) {
+				throw new InvalidDataAccessResourceUsageException("Unexpected cursor position change.");
+			}
+		}
+	}
+
+	/**
+	 * Close the cursor and database connection. Make call to cleanupOnClose so sub classes can cleanup 
+	 * any resources they have allocated.
+	 */
+	protected void doClose() throws Exception {
+		initialized = false;
+		JdbcUtils.closeResultSet(this.rs);
+		rs = null;
+		cleanupOnClose();
+		if (useSharedExtendedConnection && dataSource instanceof ExtendedConnectionDataSourceProxy) {
+			((ExtendedConnectionDataSourceProxy)dataSource).stopCloseSuppression(this.con);
+			if (!TransactionSynchronizationManager.isActualTransactionActive()) {
+				DataSourceUtils.releaseConnection(con, dataSource);
+			}
+		}
+		else {
+			JdbcUtils.closeConnection(this.con);
+		}
+	}
+	
+	protected abstract void cleanupOnClose()  throws Exception;
+
+	/**
+	 * Execute the statement to open the cursor.
+	 */
+	protected void doOpen() throws Exception {
+		Assert.state(!initialized, "Stream is already initialized.  Close before re-opening.");
+		Assert.isNull(rs, "ResultSet still open!  Close before re-opening.");
+		initializeConnection();
+		openCursor(con);
+		initialized = true;
+	
+	}
+
+	protected void initializeConnection() {
+		Assert.state(getDataSource() != null, "DataSource must not be null.");
+		
+		try {
+			if (useSharedExtendedConnection) {
+				if (!(getDataSource() instanceof ExtendedConnectionDataSourceProxy)) {
+					throw new InvalidDataAccessApiUsageException(
+							"You must use a ExtendedConnectionDataSourceProxy for the dataSource when " +
+							"useSharedExtendedConnection is set to true.");
+				}
+				this.con = DataSourceUtils.getConnection(dataSource);
+				((ExtendedConnectionDataSourceProxy)dataSource).startCloseSuppression(this.con);
+			}
+			else {
+				this.con = dataSource.getConnection();
+			}
+		}
+		catch (SQLException se) {
+			close();
+			throw getExceptionTranslator().translate("Executing query", getSql(), se);
+		}
+	}
+	
+	protected abstract void openCursor(Connection con);
+	
+	/**
+	 * Read next row and map it to item, verify cursor position if
+	 * {@link #setVerifyCursorPosition(boolean)} is true.
+	 */
+	protected T doRead() throws Exception {
+		if (rs == null) {
+			throw new ReaderNotOpenException("Reader must be open before it can be read.");
+		}
+	
+		try {
+			if (!rs.next()) {
+				return null;
+			}
+			int currentRow = getCurrentItemCount();
+			T item = readCursor(rs, currentRow);
+			verifyCursorPosition(currentRow);
+			return item;
+		}
+		catch (SQLException se) {
+			throw getExceptionTranslator().translate("Attempt to process next row failed", getSql(), se);
+		}
+	}
+	
+	/**
+	 * Read the cursor and map to the type of object this reader should return. This method must be 
+	 * overriden by subclasses.
+	 *  
+	 * @param rs The current result set
+	 * @param currentRow Current position of the result set
+	 * @return the mapped object at the cursor position
+	 * @throws SQLException
+	 */
+	protected abstract T readCursor(ResultSet rs, int currentRow) throws SQLException;
+
+	/**
+	 * Use {@link ResultSet#absolute(int)} if possible, otherwise scroll by
+	 * calling {@link ResultSet#next()}.
+	 */
+	protected void jumpToItem(int itemIndex) throws Exception {
+		if (driverSupportsAbsolute) {
+			try {
+				rs.absolute(itemIndex);
+			}
+			catch (SQLException e) {
+				// Driver does not support rs.absolute(int) revert to
+				// traversing ResultSet
+				log.warn("The JDBC driver does not appear to support ResultSet.absolute(). Consider"
+						+ " reverting to the default behavior setting the driverSupportsAbsolute to false", e);
+	
+				moveCursorToRow(itemIndex);
+			}
+		}
+		else {
+			moveCursorToRow(itemIndex);
+		}
+	}
+
+}
\ No newline at end of file
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcCursorItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcCursorItemReader.java
index 2ad554418..3236f2aee 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcCursorItemReader.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcCursorItemReader.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2007 the original author or authors.
+ * Copyright 2006-2009 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.
@@ -20,47 +20,21 @@ import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
-import java.sql.SQLWarning;
-import java.sql.Statement;
 
-import javax.sql.DataSource;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.springframework.batch.item.ExecutionContext;
-import org.springframework.batch.item.ItemStream;
-import org.springframework.batch.item.ReaderNotOpenException;
-import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader;
-import org.springframework.beans.factory.InitializingBean;
-import org.springframework.dao.InvalidDataAccessApiUsageException;
-import org.springframework.dao.InvalidDataAccessResourceUsageException;
-import org.springframework.jdbc.SQLWarningException;
 import org.springframework.jdbc.core.PreparedStatementSetter;
 import org.springframework.jdbc.core.RowMapper;
-import org.springframework.jdbc.datasource.DataSourceUtils;
 import org.springframework.jdbc.support.JdbcUtils;
-import org.springframework.jdbc.support.SQLErrorCodeSQLExceptionTranslator;
-import org.springframework.jdbc.support.SQLExceptionTranslator;
-import org.springframework.jdbc.support.SQLStateSQLExceptionTranslator;
-import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 
 /**
  * <p>
- * Simple item reader that opens a JDBC cursor and continually retrieves the
+ * Simple item reader implementation that opens a JDBC cursor and continually retrieves the
  * next row in the ResultSet. 
  * </p>
  * 
  * <p>
- * By default the cursor will be opened using a separate connection. The ResultSet for the cursor
- * is held open regardless of commits or roll backs in a surrounding transaction. Clients of this 
- * reader are responsible for buffering the items in the case that they need to be re-presented on a
- * rollback. This buffering is handled by the step implementations provided and is only a concern for
- * anyone writing their own step implementations.
- * </p>
- * 
- * <p>
  * The statement used to open the cursor is created with the 'READ_ONLY' option since a non read-only 
  * cursor may unnecessarily lock tables or rows. It is also opened with 'TYPE_FORWARD_ONLY' option. 
  * By default the cursor will be opened using a separate connection which means that it will not participate 
@@ -68,44 +42,8 @@ import org.springframework.util.ClassUtils;
  * </p>
  *  
  * <p>
- * There is an option ({@link #setUseSharedExtendedConnection(boolean)} that will share the connection 
- * used for the cursor with the rest of the step processing. If you set this flag to <code>true</code> 
- * then you must wrap the DataSource in a {@link ExtendedConnectionDataSourceProxy} to prevent the 
- * connection from being closed and released after each commit performed as part of the step processing. 
- * You must also use a JDBC driver supporting JDBC 3.0 or later since the cursor will be opened with the 
- * additional option of 'HOLD_CUSORS_OVER_COMMIT' enabled. 
- * </p>
- * 
- * <p>
  * Each call to {@link #read()} will call the provided RowMapper, passing in the
- * ResultSet. There is currently no wrapping of the ResultSet to suppress calls
- * to next(). However, if the RowMapper (mistakenly) increments the current row,
- * the next call to read will verify that the current row is at the expected
- * position and throw a DataAccessException if it is not. The reason for such strictness on the
- * ResultSet is due to the need to maintain control for transactions and
- * restartability. This ensures that each call to {@link #read()} returns the
- * ResultSet at the correct row, regardless of rollbacks or restarts.
- * </p>
- * 
- * <p>
- * {@link ExecutionContext}: The current row is returned as restart data, and
- * when restored from that same data, the cursor is opened and the current row
- * set to the value within the restart data. See
- * {@link #setDriverSupportsAbsolute(boolean)} for improving restart
- * performance.
- * </p>
- * 
- * <p>
- * Calling close on this {@link ItemStream} will cause all resources it is
- * currently using to be freed. (Connection, ResultSet, etc). It is then illegal
- * to call {@link #read()} again until it has been re-opened.
- * </p>
- * 
- * <p>
- * Known limitation: when used with Derby
- * {@link #setVerifyCursorPosition(boolean)} needs to be <code>false</code>
- * because {@link ResultSet#getRow()} call used for cursor position verification
- * is not available for 'TYPE_FORWARD_ONLY' result sets.
+ * ResultSet. 
  * </p>
  * 
  * @author Lucas Ward
@@ -113,250 +51,21 @@ import org.springframework.util.ClassUtils;
  * @author Robert Kasanicky
  * @author Thomas Risberg
  */
-public class JdbcCursorItemReader<T> extends AbstractItemCountingItemStreamItemReader<T> implements InitializingBean {
+public class JdbcCursorItemReader<T> extends AbstractCursorItemReader<T> {
 
-	private static Log log = LogFactory.getLog(JdbcCursorItemReader.class);
+	PreparedStatement preparedStatement;
 
-	public static final int VALUE_NOT_SET = -1;
+	PreparedStatementSetter preparedStatementSetter;
 
-	private Connection con;
+	String sql;
 
-	private PreparedStatement preparedStatement;
-
-	private PreparedStatementSetter preparedStatementSetter;
-
-	protected ResultSet rs;
-
-	private DataSource dataSource;
-
-	private String sql;
-
-	private int fetchSize = VALUE_NOT_SET;
-
-	private int maxRows = VALUE_NOT_SET;
-
-	private int queryTimeout = VALUE_NOT_SET;
-
-	private boolean ignoreWarnings = true;
-
-	private boolean verifyCursorPosition = true;
-
-	private SQLExceptionTranslator exceptionTranslator;
-
-	private RowMapper rowMapper;
-
-	private boolean initialized = false;
-
-	private boolean driverSupportsAbsolute = false;
-
-	private boolean useSharedExtendedConnection = false;
+	RowMapper rowMapper;
 
 	public JdbcCursorItemReader() {
+		super();
 		setName(ClassUtils.getShortName(JdbcCursorItemReader.class));
 	}
 
-	/**
-	 * Assert that mandatory properties are set.
-	 * 
-	 * @throws IllegalArgumentException if either data source or sql properties
-	 * not set.
-	 */
-	public void afterPropertiesSet() throws Exception {
-		Assert.notNull(dataSource, "DataSource must be provided");
-		Assert.notNull(sql, "The SQL query must be provided");
-		Assert.notNull(rowMapper, "RowMapper must be provided");
-	}
-
-	/**
-	 * Public setter for the data source for injection purposes.
-	 * 
-	 * @param dataSource
-	 */
-	public void setDataSource(DataSource dataSource) {
-		this.dataSource = dataSource;
-	}
-
-	/**
-	 * Executes the provided SQL query. 
-	 */
-	private void executeQuery() {
-
-		Assert.state(dataSource != null, "DataSource must not be null.");
-
-		try {
-			if (useSharedExtendedConnection) {
-				if (!(dataSource instanceof ExtendedConnectionDataSourceProxy)) {
-					throw new InvalidDataAccessApiUsageException(
-							"You must use a ExtendedConnectionDataSourceProxy for the dataSource when " +
-							"useSharedExtendedConnection is set to true.");
-				}
-				this.con = DataSourceUtils.getConnection(dataSource);
-				((ExtendedConnectionDataSourceProxy)dataSource).startCloseSuppression(this.con);
-				preparedStatement = this.con.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY,
-						ResultSet.HOLD_CURSORS_OVER_COMMIT);
-			}
-			else {
-				this.con = dataSource.getConnection();
-				preparedStatement = this.con.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
-			}
-			applyStatementSettings(preparedStatement);
-			if (this.preparedStatementSetter != null) {
-				preparedStatementSetter.setValues(preparedStatement);
-			}
-			this.rs = preparedStatement.executeQuery();
-			handleWarnings(preparedStatement);
-		}
-		catch (SQLException se) {
-			close();
-			throw getExceptionTranslator().translate("Executing query", sql, se);
-		}
-
-	}
-
-	/**
-	 * Prepare the given JDBC Statement (or PreparedStatement or
-	 * CallableStatement), applying statement settings such as fetch size, max
-	 * rows, and query timeout. @param stmt the JDBC Statement to prepare
-	 * @throws SQLException
-	 * 
-	 * @see #setFetchSize
-	 * @see #setMaxRows
-	 * @see #setQueryTimeout
-	 */
-	private void applyStatementSettings(PreparedStatement stmt) throws SQLException {
-		if (fetchSize != VALUE_NOT_SET) {
-			stmt.setFetchSize(fetchSize);
-			stmt.setFetchDirection(ResultSet.FETCH_FORWARD);
-		}
-		if (maxRows != VALUE_NOT_SET) {
-			stmt.setMaxRows(maxRows);
-		}
-		if (queryTimeout != VALUE_NOT_SET) {
-			stmt.setQueryTimeout(queryTimeout);
-		}
-	}
-
-	/**
-	 * Return the exception translator for this instance.
-	 * 
-	 * Creates a default SQLErrorCodeSQLExceptionTranslator for the specified
-	 * DataSource if none is set.
-	 */
-	protected SQLExceptionTranslator getExceptionTranslator() {
-		if (exceptionTranslator == null) {
-			if (dataSource != null) {
-				exceptionTranslator = new SQLErrorCodeSQLExceptionTranslator(dataSource);
-			}
-			else {
-				exceptionTranslator = new SQLStateSQLExceptionTranslator();
-			}
-		}
-		return exceptionTranslator;
-	}
-
-	/**
-	 * Throw a SQLWarningException if we're not ignoring warnings, else log the
-	 * warnings (at debug level).
-	 * 
-	 * @param warnings the warnings object from the current statement. May be
-	 * <code>null</code>, in which case this method does nothing.
-	 * @throws SQLException
-	 * 
-	 * @see org.springframework.jdbc.SQLWarningException
-	 */
-	private void handleWarnings(PreparedStatement pstmt) throws SQLWarningException, SQLException {
-		if (ignoreWarnings) {
-			if (log.isDebugEnabled()) {
-				SQLWarning warningToLog = pstmt.getWarnings();
-				while (warningToLog != null) {
-					log.debug("SQLWarning ignored: SQL state '" + warningToLog.getSQLState() + "', error code '"
-							+ warningToLog.getErrorCode() + "', message [" + warningToLog.getMessage() + "]");
-					warningToLog = warningToLog.getNextWarning();
-				}
-			}
-		}
-		else {
-			SQLWarning warnings = pstmt.getWarnings();
-			if (warnings != null) {
-				throw new SQLWarningException("Warning not ignored", warnings);
-			}
-		}
-	}
-
-	/**
-	 * Moves the cursor in the ResultSet to the position specified by the row
-	 * parameter by traversing the ResultSet.
-	 * @param row
-	 */
-	private void moveCursorToRow(int row) {
-		try {
-			int count = 0;
-			while (row != count && rs.next()) {
-				count++;
-			}
-		}
-		catch (SQLException se) {
-			throw getExceptionTranslator().translate("Attempted to move ResultSet to last committed row", sql, se);
-		}
-	}
-
-	/**
-	 * Gives the JDBC driver a hint as to the number of rows that should be
-	 * fetched from the database when more rows are needed for this
-	 * <code>ResultSet</code> object. If the fetch size specified is zero, the
-	 * JDBC driver ignores the value.
-	 * 
-	 * @param fetchSize the number of rows to fetch
-	 * @see ResultSet#setFetchSize(int)
-	 */
-	public void setFetchSize(int fetchSize) {
-		this.fetchSize = fetchSize;
-	}
-
-	/**
-	 * Sets the limit for the maximum number of rows that any
-	 * <code>ResultSet</code> object can contain to the given number.
-	 * 
-	 * @param maxRows the new max rows limit; zero means there is no limit
-	 * @see Statement#setMaxRows(int)
-	 */
-	public void setMaxRows(int maxRows) {
-		this.maxRows = maxRows;
-	}
-
-	/**
-	 * Sets the number of seconds the driver will wait for a
-	 * <code>Statement</code> object to execute to the given number of seconds.
-	 * If the limit is exceeded, an <code>SQLException</code> is thrown.
-	 * 
-	 * @param queryTimeout seconds the new query timeout limit in seconds; zero
-	 * means there is no limit
-	 * @see Statement#setQueryTimeout(int)
-	 */
-	public void setQueryTimeout(int queryTimeout) {
-		this.queryTimeout = queryTimeout;
-	}
-
-	/**
-	 * Set whether SQLWarnings should be ignored (only logged) or exception
-	 * should be thrown.
-	 * 
-	 * @param ignoreWarnings if TRUE, warnings are ignored
-	 */
-	public void setIgnoreWarnings(boolean ignoreWarnings) {
-		this.ignoreWarnings = ignoreWarnings;
-	}
-
-	/**
-	 * Allow verification of cursor position after current row is processed by
-	 * RowMapper or RowCallbackHandler. Default value is TRUE.
-	 * 
-	 * @param verifyCursorPosition if true, cursor position is verified
-	 */
-	public void setVerifyCursorPosition(boolean verifyCursorPosition) {
-		this.verifyCursorPosition = verifyCursorPosition;
-	}
-
 	/**
 	 * Set the RowMapper to be used for all calls to read().
 	 * 
@@ -388,127 +97,56 @@ public class JdbcCursorItemReader<T> extends AbstractItemCountingItemStreamItemR
 	}
 
 	/**
-	 * Indicate whether the JDBC driver supports setting the absolute row on a
-	 * {@link ResultSet}. It is recommended that this is set to
-	 * <code>true</code> for JDBC drivers that supports ResultSet.absolute() as
-	 * it may improve performance, especially if a step fails while working with
-	 * a large data set.
+	 * Assert that mandatory properties are set.
 	 * 
-	 * @see ResultSet#absolute(int)
-	 * 
-	 * @param driverSupportsAbsolute <code>false</code> by default
+	 * @throws IllegalArgumentException if either data source or sql properties
+	 * not set.
 	 */
-	public void setDriverSupportsAbsolute(boolean driverSupportsAbsolute) {
-		this.driverSupportsAbsolute = driverSupportsAbsolute;
+	public void afterPropertiesSet() throws Exception {
+		super.afterPropertiesSet();
+		Assert.notNull(sql, "The SQL query must be provided");
+		Assert.notNull(rowMapper, "RowMapper must be provided");
 	}
 
-	/**
-	 * Indicate whether the connection used for the cursor should be used by all other processing
-	 * thus sharing the same transaction. If this is set to false, which is the default, then the 
-	 * cursor will be opened using in its connection and will not participate in any transactions
-	 * started for the rest of the step processing. If you set this flag to true then you must 
-	 * wrap the DataSource in a {@link ExtendedConnectionDataSourceProxy} to prevent the
-	 * connection from being closed and released after each commit.
-	 * 
-	 * When you set this option to <code>true</code> then the statement used to open the cursor 
-	 * will be created with both 'READ_ONLY' and 'HOLD_CUSORS_OVER_COMMIT' options. This allows 
-	 * holding the cursor open over transaction start and commits performed in the step processing. 
-	 * To use this feature you need a database that supports this and a JDBC driver supporting 
-	 * JDBC 3.0 or later.
-	 *
-	 * @param useSharedExtendedConnection <code>false</code> by default
-	 */
-	public void setUseSharedExtendedConnection(boolean useSharedExtendedConnection) {
-		this.useSharedExtendedConnection = useSharedExtendedConnection;
-	}
 
-	/**
-	 * Check the result set is in synch with the currentRow attribute. This is
-	 * important to ensure that the user hasn't modified the current row.
-	 */
-	private void verifyCursorPosition(long expectedCurrentRow) throws SQLException {
-		if (verifyCursorPosition) {
-			if (expectedCurrentRow != this.rs.getRow()) {
-				throw new InvalidDataAccessResourceUsageException("Unexpected cursor position change.");
+	protected void openCursor(Connection con) {	
+		try {
+			if (isUseSharedExtendedConnection()) {
+				preparedStatement = con.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY,
+						ResultSet.HOLD_CURSORS_OVER_COMMIT);
 			}
+			else {
+				preparedStatement = con.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+			}
+			applyStatementSettings(preparedStatement);
+			if (this.preparedStatementSetter != null) {
+				preparedStatementSetter.setValues(preparedStatement);
+			}
+			this.rs = preparedStatement.executeQuery();
+			handleWarnings(preparedStatement);
 		}
+		catch (SQLException se) {
+			close();
+			throw getExceptionTranslator().translate("Executing query", getSql(), se);
+		}
+	
 	}
 
+
+	@SuppressWarnings("unchecked")
+	protected T readCursor(ResultSet rs, int currentRow) throws SQLException {
+		return (T) rowMapper.mapRow(rs, currentRow);
+	}
+	
 	/**
 	 * Close the cursor and database connection.
 	 */
-	protected void doClose() throws Exception {
-		initialized = false;
-		JdbcUtils.closeResultSet(this.rs);
-		rs = null;
+	protected void cleanupOnClose() throws Exception {
 		JdbcUtils.closeStatement(this.preparedStatement);
-		if (useSharedExtendedConnection && dataSource instanceof ExtendedConnectionDataSourceProxy) {
-			((ExtendedConnectionDataSourceProxy)dataSource).stopCloseSuppression(this.con);
-			if (!TransactionSynchronizationManager.isActualTransactionActive()) {
-				DataSourceUtils.releaseConnection(con, dataSource);
-			}
-		}
-		else {
-			JdbcUtils.closeConnection(this.con);
-		}
 	}
 
-	/**
-	 * Execute the {@link #setSql(String)} query.
-	 */
-	protected void doOpen() throws Exception {
-		Assert.state(!initialized, "Stream is already initialized.  Close before re-opening.");
-		Assert.isNull(rs, "ResultSet still open!  Close before re-opening.");
-		executeQuery();
-		initialized = true;
-
+	@Override
+	public String getSql() {
+		return this.sql;
 	}
-
-	/**
-	 * Read next row and map it to item, verify cursor position if
-	 * {@link #setVerifyCursorPosition(boolean)} is true.
-	 */
-	@SuppressWarnings("unchecked")
-	protected T doRead() throws Exception {
-		if (rs == null) {
-			throw new ReaderNotOpenException("Reader must be open before it can be read.");
-		}
-
-		try {
-			if (!rs.next()) {
-				return null;
-			}
-			int currentRow = getCurrentItemCount();
-			T item = (T) rowMapper.mapRow(rs, currentRow);
-			verifyCursorPosition(currentRow);
-			return item;
-		}
-		catch (SQLException se) {
-			throw getExceptionTranslator().translate("Attempt to process next row failed", sql, se);
-		}
-	}
-
-	/**
-	 * Use {@link ResultSet#absolute(int)} if possible, otherwise scroll by
-	 * calling {@link ResultSet#next()}.
-	 */
-	protected void jumpToItem(int itemIndex) throws Exception {
-		if (driverSupportsAbsolute) {
-			try {
-				rs.absolute(itemIndex);
-			}
-			catch (SQLException e) {
-				// Driver does not support rs.absolute(int) revert to
-				// traversing ResultSet
-				log.warn("The JDBC driver does not appear to support ResultSet.absolute(). Consider"
-						+ " reverting to the default behavior setting the driverSupportsAbsolute to false", e);
-
-				moveCursorToRow(itemIndex);
-			}
-		}
-		else {
-			moveCursorToRow(itemIndex);
-		}
-	}
-
 }
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/StoredProcedureItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/StoredProcedureItemReader.java
index 687d6cdf9..b9e482fb1 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/StoredProcedureItemReader.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/StoredProcedureItemReader.java
@@ -18,52 +18,24 @@ package org.springframework.batch.item.database;
 
 import java.sql.CallableStatement;
 import java.sql.Connection;
-import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
-import java.sql.SQLWarning;
-import java.sql.Statement;
 import java.sql.Types;
 
-import javax.sql.DataSource;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.springframework.batch.item.ExecutionContext;
-import org.springframework.batch.item.ItemStream;
-import org.springframework.batch.item.ReaderNotOpenException;
-import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader;
-import org.springframework.beans.factory.InitializingBean;
-import org.springframework.dao.InvalidDataAccessApiUsageException;
-import org.springframework.dao.InvalidDataAccessResourceUsageException;
-import org.springframework.jdbc.SQLWarningException;
 import org.springframework.jdbc.core.PreparedStatementSetter;
 import org.springframework.jdbc.core.RowMapper;
 import org.springframework.jdbc.core.SqlOutParameter;
 import org.springframework.jdbc.core.SqlParameter;
 import org.springframework.jdbc.core.metadata.CallMetaDataContext;
-import org.springframework.jdbc.datasource.DataSourceUtils;
 import org.springframework.jdbc.object.SqlCall;
 import org.springframework.jdbc.support.JdbcUtils;
-import org.springframework.jdbc.support.SQLErrorCodeSQLExceptionTranslator;
-import org.springframework.jdbc.support.SQLExceptionTranslator;
-import org.springframework.jdbc.support.SQLStateSQLExceptionTranslator;
-import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 
 /**
  * <p>
- * Item reader that executes a stored procedure and then reads the returned cursor and continually 
- * retrieves the next row in the ResultSet. 
- * </p>
- * 
- * <p>
- * By default the procedure will be executed using a separate connection. The ResultSet for the cursor
- * is held open regardless of commits or roll backs in a surrounding transaction. Clients of this 
- * reader are responsible for buffering the items in the case that they need to be re-presented on a
- * rollback. This buffering is handled by the step implementations provided and is only a concern for
- * anyone writing their own step implementations.
+ * Item reader implementation that executes a stored procedure and then reads the returned cursor 
+ * and continually retrieves the next row in the <code>ResultSet</code>. 
  * </p>
  * 
  * <p>
@@ -73,93 +45,28 @@ import org.springframework.util.ClassUtils;
  * </p>
  *  
  * <p>
- * There is an option ({@link #setUseSharedExtendedConnection(boolean)} that will share the connection 
- * used for the cursor with the rest of the step processing. If you set this flag to <code>true</code> 
- * then you must wrap the DataSource in a {@link ExtendedConnectionDataSourceProxy} to prevent the 
- * connection from being closed and released after each commit performed as part of the step processing. 
- * You must also use a JDBC driver supporting JDBC 3.0 or later since the cursor will be opened with the 
- * additional option of 'HOLD_CUSORS_OVER_COMMIT' enabled. 
- * </p>
- * 
- * <p>
  * Each call to {@link #read()} will call the provided RowMapper, passing in the
- * ResultSet. There is currently no wrapping of the ResultSet to suppress calls
- * to next(). However, if the RowMapper (mistakenly) increments the current row,
- * the next call to read will verify that the current row is at the expected
- * position and throw a DataAccessException if it is not. The reason for such strictness on the
- * ResultSet is due to the need to maintain control for transactions and
- * restartability. This ensures that each call to {@link #read()} returns the
- * ResultSet at the correct row, regardless of rollbacks or restarts.
- * </p>
- * 
- * <p>
- * {@link ExecutionContext}: The current row is returned as restart data, and
- * when restored from that same data, the cursor is opened and the current row
- * set to the value within the restart data. See
- * {@link #setDriverSupportsAbsolute(boolean)} for improving restart
- * performance.
- * </p>
- * 
- * <p>
- * Calling close on this {@link ItemStream} will cause all resources it is
- * currently using to be freed. (Connection, ResultSet, etc). It is then illegal
- * to call {@link #read()} again until it has been re-opened.
- * </p>
- * 
- * <p>
- * Known limitation: when used with Derby
- * {@link #setVerifyCursorPosition(boolean)} needs to be <code>false</code>
- * because {@link ResultSet#getRow()} call used for cursor position verification
- * is not available for 'TYPE_FORWARD_ONLY' result sets.
+ * ResultSet. 
  * </p>
  * 
  * <p>
  * This class is modeled after the similar <code>JdbcCursorItemReader</code> class. 
  * </p>
  * 
- * @author Lucas Ward
- * @author Peter Zozom
- * @author Robert Kasanicky
  * @author Thomas Risberg
  */
-public class StoredProcedureItemReader<T> extends AbstractItemCountingItemStreamItemReader<T> implements InitializingBean {
-
-	private static Log log = LogFactory.getLog(StoredProcedureItemReader.class);
-
-	public static final int VALUE_NOT_SET = -1;
-
-	private Connection con;
+public class StoredProcedureItemReader<T> extends AbstractCursorItemReader<T> {
 
 	private CallableStatement callableStatement;
 
 	private PreparedStatementSetter preparedStatementSetter;
 
-	protected ResultSet rs;
-
-	private DataSource dataSource;
-
 	private String procedureName;
-
-	private int fetchSize = VALUE_NOT_SET;
-
-	private int maxRows = VALUE_NOT_SET;
-
-	private int queryTimeout = VALUE_NOT_SET;
-
-	private boolean ignoreWarnings = true;
-
-	private boolean verifyCursorPosition = true;
-
-	private SQLExceptionTranslator exceptionTranslator;
+	
+	private String callString;
 
 	private RowMapper rowMapper;
 
-	private boolean initialized = false;
-
-	private boolean driverSupportsAbsolute = false;
-
-	private boolean useSharedExtendedConnection = false;
-	
 	private SqlParameter[] parameters = new SqlParameter[0];
 	
 	private boolean function = false;
@@ -167,259 +74,10 @@ public class StoredProcedureItemReader<T> extends AbstractItemCountingItemStream
 	private int refCursorPosition = 0;
 
 	public StoredProcedureItemReader() {
+		super();
 		setName(ClassUtils.getShortName(StoredProcedureItemReader.class));
 	}
 
-	/**
-	 * Assert that mandatory properties are set.
-	 * 
-	 * @throws IllegalArgumentException if either data source or sql properties
-	 * not set.
-	 */
-	public void afterPropertiesSet() throws Exception {
-		Assert.notNull(dataSource, "DataSource must be provided");
-		Assert.notNull(procedureName, "The name of the stored procedure must be provided");
-		Assert.notNull(rowMapper, "RowMapper must be provided");
-	}
-
-	/**
-	 * Public setter for the data source for injection purposes.
-	 * 
-	 * @param dataSource
-	 */
-	public void setDataSource(DataSource dataSource) {
-		this.dataSource = dataSource;
-	}
-
-	/**
-	 * Executes the provided SQL query. 
-	 */
-	private void executeQuery() {
-
-		Assert.state(dataSource != null, "DataSource must not be null.");
-		Assert.state(procedureName != null, "Procedure Name must not be null.");
-		Assert.state(refCursorPosition == 0 || refCursorPosition > parameters.length, 
-				"refCursorPosition specified as " + refCursorPosition + " but there are only " + 
-				parameters.length + " parameters defined.");
-
-		CallMetaDataContext callContext = new CallMetaDataContext();
-		callContext.setAccessCallParameterMetaData(false);
-		callContext.setProcedureName(procedureName);
-		callContext.setFunction(function);
-		callContext.initializeMetaData(dataSource);
-		SqlParameter cursorParameter = callContext.createReturnResultSetParameter("return", rowMapper);
-					
-		int cursorSqlType = Types.OTHER;
-		if (function) {
-			if (cursorParameter instanceof SqlOutParameter) {
-				cursorSqlType = cursorParameter.getSqlType();
-			}
-		}
-		else {
-			if (refCursorPosition > 0 && refCursorPosition <= parameters.length) {
-				cursorSqlType = parameters[refCursorPosition - 1].getSqlType();
-			}
-		}
-		
-		SqlCall call = new SqlCall(dataSource, procedureName){};
-		call.setParameters(parameters);
-		call.setFunction(function);
-		call.compile();
-		String callString = call.getCallString();
-		call = null;
-
-		try {
-			if (useSharedExtendedConnection) {
-				if (!(dataSource instanceof ExtendedConnectionDataSourceProxy)) {
-					throw new InvalidDataAccessApiUsageException(
-							"You must use a ExtendedConnectionDataSourceProxy for the dataSource when " +
-							"useSharedExtendedConnection is set to true.");
-				}
-				this.con = DataSourceUtils.getConnection(dataSource);
-				((ExtendedConnectionDataSourceProxy)dataSource).startCloseSuppression(this.con);
-				callableStatement = this.con.prepareCall(callString, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY,
-						ResultSet.HOLD_CURSORS_OVER_COMMIT);
-			}
-			else {
-				this.con = dataSource.getConnection();
-				callableStatement = this.con.prepareCall(callString, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
-			}
-			applyStatementSettings(callableStatement);
-			if (this.preparedStatementSetter != null) {
-				preparedStatementSetter.setValues(callableStatement);
-			}
-			
-			if (function) {
-				callableStatement.registerOutParameter(1, cursorSqlType);
-			}
-			else {
-				callableStatement.registerOutParameter(refCursorPosition, cursorSqlType);
-			}
-			boolean results = callableStatement.execute();
-			if (results) {
-				rs = callableStatement.getResultSet();
-			}
-			else {
-				if (function) {
-					rs = (ResultSet) callableStatement.getObject(1);
-				}
-				else {
-					rs = (ResultSet) callableStatement.getObject(refCursorPosition);
-				}
-			}
-			handleWarnings(callableStatement);
-		}
-		catch (SQLException se) {
-			close();
-			throw getExceptionTranslator().translate("Executing stored procedure", callString, se);
-		}
-
-	}
-
-	/**
-	 * Prepare the given JDBC Statement (or PreparedStatement or
-	 * CallableStatement), applying statement settings such as fetch size, max
-	 * rows, and query timeout. @param stmt the JDBC Statement to prepare
-	 * @throws SQLException
-	 * 
-	 * @see #setFetchSize
-	 * @see #setMaxRows
-	 * @see #setQueryTimeout
-	 */
-	private void applyStatementSettings(PreparedStatement stmt) throws SQLException {
-		if (fetchSize != VALUE_NOT_SET) {
-			stmt.setFetchSize(fetchSize);
-			stmt.setFetchDirection(ResultSet.FETCH_FORWARD);
-		}
-		if (maxRows != VALUE_NOT_SET) {
-			stmt.setMaxRows(maxRows);
-		}
-		if (queryTimeout != VALUE_NOT_SET) {
-			stmt.setQueryTimeout(queryTimeout);
-		}
-	}
-
-	/**
-	 * Return the exception translator for this instance.
-	 * 
-	 * Creates a default SQLErrorCodeSQLExceptionTranslator for the specified
-	 * DataSource if none is set.
-	 */
-	protected SQLExceptionTranslator getExceptionTranslator() {
-		if (exceptionTranslator == null) {
-			if (dataSource != null) {
-				exceptionTranslator = new SQLErrorCodeSQLExceptionTranslator(dataSource);
-			}
-			else {
-				exceptionTranslator = new SQLStateSQLExceptionTranslator();
-			}
-		}
-		return exceptionTranslator;
-	}
-
-	/**
-	 * Throw a SQLWarningException if we're not ignoring warnings, else log the
-	 * warnings (at debug level).
-	 * 
-	 * @param warnings the warnings object from the current statement. May be
-	 * <code>null</code>, in which case this method does nothing.
-	 * @throws SQLException
-	 * 
-	 * @see org.springframework.jdbc.SQLWarningException
-	 */
-	private void handleWarnings(PreparedStatement pstmt) throws SQLWarningException, SQLException {
-		if (ignoreWarnings) {
-			if (log.isDebugEnabled()) {
-				SQLWarning warningToLog = pstmt.getWarnings();
-				while (warningToLog != null) {
-					log.debug("SQLWarning ignored: SQL state '" + warningToLog.getSQLState() + "', error code '"
-							+ warningToLog.getErrorCode() + "', message [" + warningToLog.getMessage() + "]");
-					warningToLog = warningToLog.getNextWarning();
-				}
-			}
-		}
-		else {
-			SQLWarning warnings = pstmt.getWarnings();
-			if (warnings != null) {
-				throw new SQLWarningException("Warning not ignored", warnings);
-			}
-		}
-	}
-
-	/**
-	 * Moves the cursor in the ResultSet to the position specified by the row
-	 * parameter by traversing the ResultSet.
-	 * @param row
-	 */
-	private void moveCursorToRow(int row) {
-		try {
-			int count = 0;
-			while (row != count && rs.next()) {
-				count++;
-			}
-		}
-		catch (SQLException se) {
-			throw getExceptionTranslator().translate("Attempted to move ResultSet to last committed row", procedureName, se);
-		}
-	}
-
-	/**
-	 * Gives the JDBC driver a hint as to the number of rows that should be
-	 * fetched from the database when more rows are needed for this
-	 * <code>ResultSet</code> object. If the fetch size specified is zero, the
-	 * JDBC driver ignores the value.
-	 * 
-	 * @param fetchSize the number of rows to fetch
-	 * @see ResultSet#setFetchSize(int)
-	 */
-	public void setFetchSize(int fetchSize) {
-		this.fetchSize = fetchSize;
-	}
-
-	/**
-	 * Sets the limit for the maximum number of rows that any
-	 * <code>ResultSet</code> object can contain to the given number.
-	 * 
-	 * @param maxRows the new max rows limit; zero means there is no limit
-	 * @see Statement#setMaxRows(int)
-	 */
-	public void setMaxRows(int maxRows) {
-		this.maxRows = maxRows;
-	}
-
-	/**
-	 * Sets the number of seconds the driver will wait for a
-	 * <code>Statement</code> object to execute to the given number of seconds.
-	 * If the limit is exceeded, an <code>SQLException</code> is thrown.
-	 * 
-	 * @param queryTimeout seconds the new query timeout limit in seconds; zero
-	 * means there is no limit
-	 * @see Statement#setQueryTimeout(int)
-	 */
-	public void setQueryTimeout(int queryTimeout) {
-		this.queryTimeout = queryTimeout;
-	}
-
-	/**
-	 * Set whether SQLWarnings should be ignored (only logged) or exception
-	 * should be thrown.
-	 * 
-	 * @param ignoreWarnings if TRUE, warnings are ignored
-	 */
-	public void setIgnoreWarnings(boolean ignoreWarnings) {
-		this.ignoreWarnings = ignoreWarnings;
-	}
-
-	/**
-	 * Allow verification of cursor position after current row is processed by
-	 * RowMapper or RowCallbackHandler. Default value is TRUE.
-	 * 
-	 * @param verifyCursorPosition if true, cursor position is verified
-	 */
-	public void setVerifyCursorPosition(boolean verifyCursorPosition) {
-		this.verifyCursorPosition = verifyCursorPosition;
-	}
-
 	/**
 	 * Set the RowMapper to be used for all calls to read().
 	 * 
@@ -450,41 +108,6 @@ public class StoredProcedureItemReader<T> extends AbstractItemCountingItemStream
 		this.preparedStatementSetter = preparedStatementSetter;
 	}
 
-	/**
-	 * Indicate whether the JDBC driver supports setting the absolute row on a
-	 * {@link ResultSet}. It is recommended that this is set to
-	 * <code>true</code> for JDBC drivers that supports ResultSet.absolute() as
-	 * it may improve performance, especially if a step fails while working with
-	 * a large data set.
-	 * 
-	 * @see ResultSet#absolute(int)
-	 * 
-	 * @param driverSupportsAbsolute <code>false</code> by default
-	 */
-	public void setDriverSupportsAbsolute(boolean driverSupportsAbsolute) {
-		this.driverSupportsAbsolute = driverSupportsAbsolute;
-	}
-
-	/**
-	 * Indicate whether the connection used for the cursor should be used by all other processing
-	 * thus sharing the same transaction. If this is set to false, which is the default, then the 
-	 * cursor will be opened using in its connection and will not participate in any transactions
-	 * started for the rest of the step processing. If you set this flag to true then you must 
-	 * wrap the DataSource in a {@link ExtendedConnectionDataSourceProxy} to prevent the
-	 * connection from being closed and released after each commit.
-	 * 
-	 * When you set this option to <code>true</code> then the statement used to open the cursor 
-	 * will be created with both 'READ_ONLY' and 'HOLD_CUSORS_OVER_COMMIT' options. This allows 
-	 * holding the cursor open over transaction start and commits performed in the step processing. 
-	 * To use this feature you need a database that supports this and a JDBC driver supporting 
-	 * JDBC 3.0 or later.
-	 *
-	 * @param useSharedExtendedConnection <code>false</code> by default
-	 */
-	public void setUseSharedExtendedConnection(boolean useSharedExtendedConnection) {
-		this.useSharedExtendedConnection = useSharedExtendedConnection;
-	}
-	
 	/**
 	 * Add one or more declared parameters. Used for configuring this operation when used in a 
 	 * bean factory. Each parameter will specify SQL type and (optionally) the parameter's name. 
@@ -514,91 +137,109 @@ public class StoredProcedureItemReader<T> extends AbstractItemCountingItemStream
 	}
 
 	/**
-	 * Check the result set is in synch with the currentRow attribute. This is
-	 * important to ensure that the user hasn't modified the current row.
+	 * Assert that mandatory properties are set.
+	 * 
+	 * @throws IllegalArgumentException if either data source or sql properties
+	 * not set.
 	 */
-	private void verifyCursorPosition(long expectedCurrentRow) throws SQLException {
-		if (verifyCursorPosition) {
-			if (expectedCurrentRow != this.rs.getRow()) {
-				throw new InvalidDataAccessResourceUsageException("Unexpected cursor position change.");
+	public void afterPropertiesSet() throws Exception {
+		super.afterPropertiesSet();
+		Assert.notNull(procedureName, "The name of the stored procedure must be provided");
+		Assert.notNull(rowMapper, "RowMapper must be provided");
+	}
+
+	protected void openCursor(Connection con) {	
+
+		Assert.state(procedureName != null, "Procedure Name must not be null.");
+		Assert.state(refCursorPosition == 0 || refCursorPosition > parameters.length, 
+				"refCursorPosition specified as " + refCursorPosition + " but there are only " + 
+				parameters.length + " parameters defined.");
+
+		CallMetaDataContext callContext = new CallMetaDataContext();
+		callContext.setAccessCallParameterMetaData(false);
+		callContext.setProcedureName(procedureName);
+		callContext.setFunction(function);
+		callContext.initializeMetaData(getDataSource());
+		SqlParameter cursorParameter = callContext.createReturnResultSetParameter("return", rowMapper);
+					
+		int cursorSqlType = Types.OTHER;
+		if (function) {
+			if (cursorParameter instanceof SqlOutParameter) {
+				cursorSqlType = cursorParameter.getSqlType();
 			}
 		}
+		else {
+			if (refCursorPosition > 0 && refCursorPosition <= parameters.length) {
+				cursorSqlType = parameters[refCursorPosition - 1].getSqlType();
+			}
+		}
+		
+		SqlCall call = new SqlCall(getDataSource(), procedureName){};
+		call.setParameters(parameters);
+		call.setFunction(function);
+		call.compile();
+		this.callString = call.getCallString();
+		call = null;
+
+		try {
+			if (isUseSharedExtendedConnection()) {
+				callableStatement = con.prepareCall(callString, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY,
+						ResultSet.HOLD_CURSORS_OVER_COMMIT);
+			}
+			else {
+				callableStatement = con.prepareCall(callString, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+			}
+			applyStatementSettings(callableStatement);
+			if (this.preparedStatementSetter != null) {
+				preparedStatementSetter.setValues(callableStatement);
+			}
+			
+			if (function) {
+				callableStatement.registerOutParameter(1, cursorSqlType);
+			}
+			else {
+				callableStatement.registerOutParameter(refCursorPosition, cursorSqlType);
+			}
+			boolean results = callableStatement.execute();
+			if (results) {
+				rs = callableStatement.getResultSet();
+			}
+			else {
+				if (function) {
+					rs = (ResultSet) callableStatement.getObject(1);
+				}
+				else {
+					rs = (ResultSet) callableStatement.getObject(refCursorPosition);
+				}
+			}
+			handleWarnings(callableStatement);
+		}
+		catch (SQLException se) {
+			close();
+			throw getExceptionTranslator().translate("Executing stored procedure", getSql(), se);
+		}
+
+	}
+
+	@SuppressWarnings("unchecked")
+	protected T readCursor(ResultSet rs, int currentRow) throws SQLException {
+		return (T) rowMapper.mapRow(rs, currentRow);
 	}
 
 	/**
 	 * Close the cursor and database connection.
 	 */
-	protected void doClose() throws Exception {
-		initialized = false;
-		JdbcUtils.closeResultSet(this.rs);
-		rs = null;
+	protected void cleanupOnClose() throws Exception {
 		JdbcUtils.closeStatement(this.callableStatement);
-		if (useSharedExtendedConnection && dataSource instanceof ExtendedConnectionDataSourceProxy) {
-			((ExtendedConnectionDataSourceProxy)dataSource).stopCloseSuppression(this.con);
-			if (!TransactionSynchronizationManager.isActualTransactionActive()) {
-				DataSourceUtils.releaseConnection(con, dataSource);
-			}
+	}
+
+	@Override
+	public String getSql() {
+		if (callString != null) {
+			return this.callString;
 		}
 		else {
-			JdbcUtils.closeConnection(this.con);
-		}
-	}
-
-	/**
-	 * Execute the {@link #setSql(String)} query.
-	 */
-	protected void doOpen() throws Exception {
-		Assert.state(!initialized, "Stream is already initialized.  Close before re-opening.");
-		Assert.isNull(rs, "ResultSet still open!  Close before re-opening.");
-		executeQuery();
-		initialized = true;
-
-	}
-
-	/**
-	 * Read next row and map it to item, verify cursor position if
-	 * {@link #setVerifyCursorPosition(boolean)} is true.
-	 */
-	@SuppressWarnings("unchecked")
-	protected T doRead() throws Exception {
-		if (rs == null) {
-			throw new ReaderNotOpenException("Reader must be open before it can be read.");
-		}
-
-		try {
-			if (!rs.next()) {
-				return null;
-			}
-			int currentRow = getCurrentItemCount();
-			T item = (T) rowMapper.mapRow(rs, currentRow);
-			verifyCursorPosition(currentRow);
-			return item;
-		}
-		catch (SQLException se) {
-			throw getExceptionTranslator().translate("Attempt to process next row failed", procedureName, se);
-		}
-	}
-
-	/**
-	 * Use {@link ResultSet#absolute(int)} if possible, otherwise scroll by
-	 * calling {@link ResultSet#next()}.
-	 */
-	protected void jumpToItem(int itemIndex) throws Exception {
-		if (driverSupportsAbsolute) {
-			try {
-				rs.absolute(itemIndex);
-			}
-			catch (SQLException e) {
-				// Driver does not support rs.absolute(int) revert to
-				// traversing ResultSet
-				log.warn("The JDBC driver does not appear to support ResultSet.absolute(). Consider"
-						+ " reverting to the default behavior setting the driverSupportsAbsolute to false", e);
-
-				moveCursorToRow(itemIndex);
-			}
-		}
-		else {
-			moveCursorToRow(itemIndex);
+			return "PROCEDURE NAME: " + procedureName;
 		}
 	}