From 8326f540a739011b96343a787e4d52ba7b1b5a4e Mon Sep 17 00:00:00 2001 From: robokaso Date: Thu, 17 Apr 2008 09:27:34 +0000 Subject: [PATCH] RESOLVED - BATCH-577: JdbcCursorItemReader doesn't work with Derby 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 --- .../item/database/JdbcCursorItemReader.java | 247 ++++++++++-------- 1 file changed, 143 insertions(+), 104 deletions(-) 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 38fafbedf..bb85abcd0 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 @@ -53,52 +53,70 @@ import org.springframework.util.ClassUtils; /** *

- * 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. *

* *

- * 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. *

* *

- * {@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. *

* *

- * 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. *

* *

- * 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. *

* *

- * 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. + *

+ * + *

+ * Known limitation: when used with Derby + * {@link #setVerifyCursorPosition(boolean)} needs to be false + * because {@link ResultSet#getRow()} call used for cursor position verification + * throws an exception. *

* * @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.

Creates a default SQLErrorCodeSQLExceptionTranslator for - * the specified DataSource if none is set. + * 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 { + } + 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 null, in which case this - * method does nothing. + * @param warning the warnings object from the current statement. May be + * null, 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 ResultSet 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 + * ResultSet 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 ResultSet object can contain to the given - * number. + * Sets the limit for the maximum number of rows that any + * ResultSet 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 Statement object to execute to the given - * number of seconds. If the limit is exceeded, an SQLException is thrown. + * Sets the number of seconds the driver will wait for a + * Statement object to execute to the given number of + * seconds. If the limit is exceeded, an SQLException 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; }