Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

RESOLVED - BATCH-577: JdbcCursorItemReader doesn't work with Derby

Browse Source 
isolated the problematic ResultSet#getRow() call to be used only for cursor position verification and added javadoc to disable the verification when used with Derby
This commit is contained in:
robokaso
2008-04-17 09:27:34 +00:00
parent 73227ce6e6
commit 8326f540a7
1 changed files with 143 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

247
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcCursorItemReader.java
View File

@@ -53,52 +53,70 @@ import org.springframework.util.ClassUtils;
/**
* <p>
* Simple input source that opens a JDBC cursor and continually retrieves the next row in the ResultSet. It is extremely
* important to note that the JdbcDriver used must be version 3.0 or higher. This is because earlier versions do not
* support holding a ResultSet open over commits.
* Simple input source that opens a JDBC cursor and continually retrieves the
* next row in the ResultSet. It is extremely important to note that the
* JdbcDriver used must be version 3.0 or higher. This is because earlier
* versions do not support holding a ResultSet open over commits.
* </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. This means that, in theory, a RowMapper could read ahead, as long as it returns the
* row back to the correct position before returning. The reason for such strictness on the ResultSet is due to the need
* to maintain control for transactions, restartability and skippability. This ensures that each call to {@link #read()}
* returns the ResultSet at the correct line, regardless of rollbacks, restarts, or skips.
* 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. This means that, in
* theory, a RowMapper could read ahead, as long as it returns the row back to
* the correct position before returning. The reason for such strictness on the
* ResultSet is due to the need to maintain control for transactions,
* restartability and skippability. This ensures that each call to
* {@link #read()} returns the ResultSet at the correct line, regardless of
* rollbacks, restarts, or skips.
* </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. Two values are stored: the current
* {@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. Two values are stored: the current
* line being processed and the number of lines that have been skipped.
* </p>
*
* <p>
* Transactions: The same ResultSet is held open regardless of commits or roll backs in a surrounding transaction. This
* means that when such a transaction is committed, the input source is notified through the {@link #mark()} and
* {@link #reset()} so that it can save it's current row number. Later, if the transaction is rolled back, the current
* row can be moved back to the same row number as it was on when commit was called.
* Transactions: The same ResultSet is held open regardless of commits or roll
* backs in a surrounding transaction. This means that when such a transaction
* is committed, the input source is notified through the {@link #mark()} and
* {@link #reset()} so that it can save it's current row number. Later, if the
* transaction is rolled back, the current row can be moved back to the same row
* number as it was on when commit was called.
* </p>
*
* <p>
* Calling skip will indicate that a record is bad and should not be re-presented to the user if the transaction is
* rolled back. For example, if row 2 is read in, and found to be bad, calling skip will inform the {@link ItemReader}.
* If reading is then continued, and a rollback is necessary because of an error on output, the input source will be
* returned to row 1. Calling read while on row 1 will move the current row to 3, not 2, because 2 has been marked as
* skipped.
* Calling skip will indicate that a record is bad and should not be
* re-presented to the user if the transaction is rolled back. For example, if
* row 2 is read in, and found to be bad, calling skip will inform the
* {@link ItemReader}. If reading is then continued, and a rollback is
* necessary because of an error on output, the input source will be returned to
* row 1. Calling read while on row 1 will move the current row to 3, not 2,
* because 2 has been marked as skipped.
* </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 opened.
* 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 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
* throws an exception.
* </p>
*
* @author Lucas Ward
* @author Peter Zozom
*/
public class JdbcCursorItemReader extends ExecutionContextUserSupport implements ItemReader, InitializingBean,
ItemStream {
ItemStream {
private static Log log = LogFactory.getLog(JdbcCursorItemReader.class);
@@ -108,9 +126,9 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
private Connection con;
private PreparedStatement preparedStatement;
private PreparedStatementSetter preparedStatementSetter;
private PreparedStatement preparedStatement;
private PreparedStatementSetter preparedStatementSetter;
protected ResultSet rs;
@@ -130,13 +148,12 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
private SQLExceptionTranslator exceptionTranslator;
private RowMapper mapper;
private boolean initialized = false;
private boolean saveState = false;
private BufferredResultSetReader bufferredReader;
public JdbcCursorItemReader() {
@@ -146,7 +163,8 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
/**
* Assert that mandatory properties are set.
*
* @throws IllegalArgumentException if either data source or sql properties not set.
* @throws IllegalArgumentException if either data source or sql properties
* not set.
*/
public void afterPropertiesSet() throws Exception {
Assert.notNull(dataSource, "DataSOurce must be provided");
@@ -164,16 +182,17 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
/**
* Increment the cursor to the next row, validating the cursor position and passing the resultset to the RowMapper.
* If read has not been called on this instance before, the cursor will be opened. If there are skipped records for
* this commit scope, an internal list of skipped records will be checked to ensure that only a valid row is given
* to the mapper.
* Increment the cursor to the next row, validating the cursor position and
* passing the resultset to the RowMapper. If read has not been called on
* this instance before, the cursor will be opened. If there are skipped
* records for this commit scope, an internal list of skipped records will
* be checked to ensure that only a valid row is given to the mapper.
*
* @returns Object returned by RowMapper
* @throws DataAccessException
* @throws IllegalStateExceptino if mapper is null.
*/
public Object read() throws Exception{
public Object read() throws Exception {
return bufferredReader.read();
}
@@ -183,7 +202,8 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
/**
* Mark the current row. Calling reset will cause the result set to be set to the current row when mark was called.
* Mark the current row. Calling reset will cause the result set to be set
* to the current row when mark was called.
*/
public void mark() {
bufferredReader.mark();
@@ -199,8 +219,9 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
/**
* Close this input source. The ResultSet, Statement and Connection created will be closed. This must be called or
* the connection and cursor will be held open indefinitely!
* Close this input source. The ResultSet, Statement and Connection created
* will be closed. This must be called or the connection and cursor will be
* held open indefinitely!
*
* @see org.springframework.batch.item.ResourceLifecycle#close(ExecutionContext)
*/
@@ -212,12 +233,12 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
bufferredReader = null;
}
/*
* Executes the provided SQL query. The statement is created with 'READ_ONLY' and 'HOLD_CUSORS_OVER_COMMIT' set to
* true. This is extremely important, since a non read-only cursor may lock tables that shouldn't be locked, and not
* holding the cursor open over a commit would require it to be reopened after each commit, which would destroy
* performance.
* Executes the provided SQL query. The statement is created with
* 'READ_ONLY' and 'HOLD_CUSORS_OVER_COMMIT' set to true. This is extremely
* important, since a non read-only cursor may lock tables that shouldn't be
* locked, and not holding the cursor open over a commit would require it to
* be reopened after each commit, which would destroy performance.
*/
private void executeQuery() {
@@ -226,14 +247,15 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
try {
this.con = dataSource.getConnection();
preparedStatement = this.con.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY,
ResultSet.HOLD_CURSORS_OVER_COMMIT);
ResultSet.HOLD_CURSORS_OVER_COMMIT);
applyStatementSettings(preparedStatement);
if(this.preparedStatementSetter != null){
if (this.preparedStatementSetter != null) {
preparedStatementSetter.setValues(preparedStatement);
}
this.rs = preparedStatement.executeQuery();
handleWarnings(preparedStatement.getWarnings());
} catch (SQLException se) {
}
catch (SQLException se) {
close(null);
throw getExceptionTranslator().translate("Executing query", sql, se);
}
@@ -241,8 +263,10 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
/*
* 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
* 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
@@ -262,14 +286,16 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
/*
* Return the exception translator for this instance. <p>Creates a default SQLErrorCodeSQLExceptionTranslator for
* the specified DataSource if none is set.
* Return the exception translator for this instance. <p>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 {
}
else {
exceptionTranslator = new SQLStateSQLExceptionTranslator();
}
}
@@ -277,10 +303,11 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
/*
* Throw a SQLWarningException if we're not ignoring warnings, else log the warnings (at debug level).
* Throw a SQLWarningException if we're not ignoring warnings, else log the
* warnings (at debug level).
*
* @param warning the warnings object from the current statement. May be <code>null</code>, in which case this
* method does nothing.
* @param warning the warnings object from the current statement. May be
* <code>null</code>, in which case this method does nothing.
*
* @see org.springframework.jdbc.SQLWarningException
*/
@@ -289,10 +316,11 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
SQLWarning warningToLog = warnings;
while (warningToLog != null) {
log.debug("SQLWarning ignored: SQL state '" + warningToLog.getSQLState() + "', error code '"
+ warningToLog.getErrorCode() + "', message [" + warningToLog.getMessage() + "]");
+ warningToLog.getErrorCode() + "', message [" + warningToLog.getMessage() + "]");
warningToLog = warningToLog.getNextWarning();
}
} else if (warnings != null) {
}
else if (warnings != null) {
throw new SQLWarningException("Warning not ignored", warnings);
}
}
@@ -320,28 +348,32 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
Assert.notNull(context, "ExecutionContext must not be null");
executeQuery();
initialized = true;
long processedRowCount = 0;
int processedRowCount = 0;
if (context.containsKey(getKey(CURRENT_PROCESSED_ROW))) {
try {
processedRowCount = context.getLong(getKey(CURRENT_PROCESSED_ROW));
while(rs.next()){
if(rs.getRow() == processedRowCount){
processedRowCount = Long.valueOf(context.getLong(getKey(CURRENT_PROCESSED_ROW))).intValue();
int count = 0;
while (rs.next()) {
count++;
if (count == processedRowCount) {
break;
}
}
} catch (SQLException se) {
}
catch (SQLException se) {
throw getExceptionTranslator().translate("Attempted to move ResultSet to last committed row", sql, se);
}
}
bufferredReader = new BufferredResultSetReader(rs, mapper, processedRowCount);
}
/**
* 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.
* 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)
@@ -351,8 +383,8 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
/**
* Sets the limit for the maximum number of rows that any <code>ResultSet</code> object can contain to the given
* number.
* 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)
@@ -362,10 +394,13 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
/**
* 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.
* 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
* @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) {
@@ -373,7 +408,8 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
/**
* Set whether SQLWarnings should be ignored (only logged) or exception should be thrown.
* Set whether SQLWarnings should be ignored (only logged) or exception
* should be thrown.
*
* @param ignoreWarnings if TRUE, warnings are ignored
*/
@@ -382,8 +418,8 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
/**
* Allow verification of cursor position after current row is processed by RowMapper or RowCallbackHandler. Default
* value is TRUE.
* 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
*/
@@ -401,23 +437,23 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
/**
* Set the sql statement to be used when creating the cursor. This statement should be a complete and valid Sql
* statement, as it will be run directly without any modification.
* Set the sql statement to be used when creating the cursor. This statement
* should be a complete and valid Sql statement, as it will be run directly
* without any modification.
*
* @param sql
*/
public void setSql(String sql) {
this.sql = sql;
}
/**
* Set the PreparedStatementSetter to use if any parameter values that need to be set in the supplied
* query.
* Set the PreparedStatementSetter to use if any parameter values that need
* to be set in the supplied query.
*
* @param preparedStatementSetter
*/
public void setPreparedStatementSetter(
PreparedStatementSetter preparedStatementSetter) {
public void setPreparedStatementSetter(PreparedStatementSetter preparedStatementSetter) {
this.preparedStatementSetter = preparedStatementSetter;
}
@@ -430,20 +466,22 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
public void setSaveState(boolean saveState) {
this.saveState = saveState;
}
private class BufferredResultSetReader implements ItemReader{
private class BufferredResultSetReader implements ItemReader {
private ResultSet rs;
private RowMapper rowMapper;
private List buffer;
private int currentIndex;
private long processedRowCount;
private int processedRowCount;
private int INITIAL_POSITION = -1;
public BufferredResultSetReader(ResultSet rs, RowMapper rowMapper, long processedRowCount) {
public BufferredResultSetReader(ResultSet rs, RowMapper rowMapper, int processedRowCount) {
Assert.notNull(rs, "The ResultSet must not be null");
Assert.notNull(rowMapper, "The RowMapper must not be null");
this.rs = rs;
@@ -452,34 +490,34 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
currentIndex = INITIAL_POSITION;
this.processedRowCount = processedRowCount;
}
public BufferredResultSetReader(ResultSet rs, RowMapper rowMapper){
public BufferredResultSetReader(ResultSet rs, RowMapper rowMapper) {
this(rs, rowMapper, 0);
}
public Object read() throws Exception, UnexpectedInputException,
NoWorkFoundException, ParseException {
public Object read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException {
currentIndex++;
// if the incremented index reaches out of the buffer, add next item from result set to buffer
if(buffer.size() == currentIndex){
try{
if(!rs.next()){
// if the incremented index reaches out of the buffer, add next item
// from result set to buffer
if (buffer.size() == currentIndex) {
try {
if (!rs.next()) {
return null;
}
int currentRow = rs.getRow();
int currentRow = processedRowCount + 1;// rs.getRow();
buffer.add(rowMapper.mapRow(rs, currentRow));
verifyCursorPosition(currentRow);
}
catch(SQLException se){
catch (SQLException se) {
throw getExceptionTranslator().translate("Attempt to process next row failed", sql, se);
}
}
processedRowCount++;
return buffer.get(currentIndex);
}
public void mark() throws MarkFailedException {
buffer.clear();
currentIndex = INITIAL_POSITION;
@@ -489,8 +527,9 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
processedRowCount -= buffer.size();
currentIndex = INITIAL_POSITION;
}
// Check the result set is in synch with the currentRow attribute. This is
// 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 {
@@ -500,7 +539,7 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
}
}
public long getProcessedRowCount() {
return processedRowCount;
}