View Javadoc

1   /*
2    * Copyright 2006-2013 the original author or authors.
3    *
4    * Licensed under the Apache License, Version 2.0 (the "License");
5    * you may not use this file except in compliance with the License.
6    * You may obtain a copy of the License at
7    *
8    *      http://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software
11   * distributed under the License is distributed on an "AS IS" BASIS,
12   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13   * See the License for the specific language governing permissions and
14   * limitations under the License.
15   */
16  
17  package org.springframework.batch.item.database;
18  
19  import java.sql.Connection;
20  import java.sql.PreparedStatement;
21  import java.sql.ResultSet;
22  import java.sql.SQLException;
23  import java.sql.SQLWarning;
24  import java.sql.Statement;
25  
26  import javax.sql.DataSource;
27  
28  import org.apache.commons.logging.Log;
29  import org.apache.commons.logging.LogFactory;
30  import org.springframework.batch.item.ExecutionContext;
31  import org.springframework.batch.item.ItemStream;
32  import org.springframework.batch.item.ReaderNotOpenException;
33  import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader;
34  import org.springframework.beans.factory.InitializingBean;
35  import org.springframework.dao.InvalidDataAccessApiUsageException;
36  import org.springframework.dao.InvalidDataAccessResourceUsageException;
37  import org.springframework.jdbc.SQLWarningException;
38  import org.springframework.jdbc.datasource.DataSourceUtils;
39  import org.springframework.jdbc.support.JdbcUtils;
40  import org.springframework.jdbc.support.SQLErrorCodeSQLExceptionTranslator;
41  import org.springframework.jdbc.support.SQLExceptionTranslator;
42  import org.springframework.jdbc.support.SQLStateSQLExceptionTranslator;
43  import org.springframework.transaction.support.TransactionSynchronizationManager;
44  import org.springframework.util.Assert;
45  
46  /**
47   * <p>
48   * Abstract base class for any simple item reader that opens a database cursor and continually retrieves
49   * the next row in the ResultSet.
50   * </p>
51   *
52   * <p>
53   * By default the cursor will be opened using a separate connection. The ResultSet for the cursor
54   * is held open regardless of commits or roll backs in a surrounding transaction. Clients of this
55   * reader are responsible for buffering the items in the case that they need to be re-presented on a
56   * rollback. This buffering is handled by the step implementations provided and is only a concern for
57   * anyone writing their own step implementations.
58   * </p>
59   *
60   * <p>
61   * There is an option ({@link #setUseSharedExtendedConnection(boolean)} that will share the connection
62   * used for the cursor with the rest of the step processing. If you set this flag to <code>true</code>
63   * then you must wrap the DataSource in a {@link ExtendedConnectionDataSourceProxy} to prevent the
64   * connection from being closed and released after each commit performed as part of the step processing.
65   * You must also use a JDBC driver supporting JDBC 3.0 or later since the cursor will be opened with the
66   * additional option of 'HOLD_CUSORS_OVER_COMMIT' enabled.
67   * </p>
68   *
69   * <p>
70   * Each call to {@link #read()} will attempt to map the row at the current position in the
71   * ResultSet. There is currently no wrapping of the ResultSet to suppress calls
72   * to next(). However, if the RowMapper (mistakenly) increments the current row,
73   * the next call to read will verify that the current row is at the expected
74   * position and throw a DataAccessException if it is not. The reason for such strictness on the
75   * ResultSet is due to the need to maintain control for transactions and
76   * restartability. This ensures that each call to {@link #read()} returns the
77   * ResultSet at the correct row, regardless of rollbacks or restarts.
78   * </p>
79   *
80   * <p>
81   * {@link ExecutionContext}: The current row is returned as restart data, and
82   * when restored from that same data, the cursor is opened and the current row
83   * set to the value within the restart data. See
84   * {@link #setDriverSupportsAbsolute(boolean)} for improving restart
85   * performance.
86   * </p>
87   *
88   * <p>
89   * Calling close on this {@link ItemStream} will cause all resources it is
90   * currently using to be freed. (Connection, ResultSet, etc). It is then illegal
91   * to call {@link #read()} again until it has been re-opened.
92   * </p>
93   *
94   * <p>
95   * Known limitation: when used with Derby
96   * {@link #setVerifyCursorPosition(boolean)} needs to be <code>false</code>
97   * because {@link ResultSet#getRow()} call used for cursor position verification
98   * is not available for 'TYPE_FORWARD_ONLY' result sets.
99   * </p>
100  *
101  * @author Lucas Ward
102  * @author Peter Zozom
103  * @author Robert Kasanicky
104  * @author Thomas Risberg
105  */
106 public abstract class AbstractCursorItemReader<T> extends AbstractItemCountingItemStreamItemReader<T>
107 implements InitializingBean {
108 
109 	/** Logger available to subclasses */
110 	protected final Log log = LogFactory.getLog(getClass());
111 
112 	public static final int VALUE_NOT_SET = -1;
113 	private Connection con;
114 
115 	protected ResultSet rs;
116 
117 	private DataSource dataSource;
118 
119 	private int fetchSize = VALUE_NOT_SET;
120 
121 	private int maxRows = VALUE_NOT_SET;
122 
123 	private int queryTimeout = VALUE_NOT_SET;
124 
125 	private boolean ignoreWarnings = true;
126 
127 	private boolean verifyCursorPosition = true;
128 
129 	private SQLExceptionTranslator exceptionTranslator;
130 
131 	private boolean initialized = false;
132 
133 	private boolean driverSupportsAbsolute = false;
134 
135 	private boolean useSharedExtendedConnection = false;
136 
137 
138 	public AbstractCursorItemReader() {
139 		super();
140 	}
141 
142 	/**
143 	 * Assert that mandatory properties are set.
144 	 *
145 	 * @throws IllegalArgumentException if either data source or sql properties
146 	 * not set.
147 	 */
148 	@Override
149 	public void afterPropertiesSet() throws Exception {
150 		Assert.notNull(dataSource, "DataSource must be provided");
151 	}
152 
153 	/**
154 	 * Public setter for the data source for injection purposes.
155 	 *
156 	 * @param dataSource
157 	 */
158 	public void setDataSource(DataSource dataSource) {
159 		this.dataSource = dataSource;
160 	}
161 
162 	/**
163 	 * Public getter for the data source.
164 	 *
165 	 * @return the dataSource
166 	 */
167 	public DataSource getDataSource() {
168 		return this.dataSource;
169 	}
170 
171 	/**
172 	 * Prepare the given JDBC Statement (or PreparedStatement or
173 	 * CallableStatement), applying statement settings such as fetch size, max
174 	 * rows, and query timeout. @param stmt the JDBC Statement to prepare
175 	 * @throws SQLException
176 	 *
177 	 * @see #setFetchSize
178 	 * @see #setMaxRows
179 	 * @see #setQueryTimeout
180 	 */
181 	protected void applyStatementSettings(PreparedStatement stmt) throws SQLException {
182 		if (fetchSize != VALUE_NOT_SET) {
183 			stmt.setFetchSize(fetchSize);
184 			stmt.setFetchDirection(ResultSet.FETCH_FORWARD);
185 		}
186 		if (maxRows != VALUE_NOT_SET) {
187 			stmt.setMaxRows(maxRows);
188 		}
189 		if (queryTimeout != VALUE_NOT_SET) {
190 			stmt.setQueryTimeout(queryTimeout);
191 		}
192 	}
193 
194 	/**
195 	 * Return the exception translator for this instance.
196 	 *
197 	 * Creates a default SQLErrorCodeSQLExceptionTranslator for the specified
198 	 * DataSource if none is set.
199 	 */
200 	protected SQLExceptionTranslator getExceptionTranslator() {
201 		synchronized(this) {
202 			if (exceptionTranslator == null) {
203 				if (dataSource != null) {
204 					exceptionTranslator = new SQLErrorCodeSQLExceptionTranslator(dataSource);
205 				}
206 				else {
207 					exceptionTranslator = new SQLStateSQLExceptionTranslator();
208 				}
209 			}
210 		}
211 		return exceptionTranslator;
212 	}
213 
214 	/**
215 	 * Throw a SQLWarningException if we're not ignoring warnings, else log the
216 	 * warnings (at debug level).
217 	 *
218 	 * @param statement the current statement to obtain the warnings from, if there are any.
219 	 * @throws SQLException
220 	 *
221 	 * @see org.springframework.jdbc.SQLWarningException
222 	 */
223 	protected void handleWarnings(Statement statement) throws SQLWarningException,
224 	SQLException {
225 		if (ignoreWarnings) {
226 			if (log.isDebugEnabled()) {
227 				SQLWarning warningToLog = statement.getWarnings();
228 				while (warningToLog != null) {
229 					log.debug("SQLWarning ignored: SQL state '" + warningToLog.getSQLState() + "', error code '"
230 							+ warningToLog.getErrorCode() + "', message [" + warningToLog.getMessage() + "]");
231 					warningToLog = warningToLog.getNextWarning();
232 				}
233 			}
234 		}
235 		else {
236 			SQLWarning warnings = statement.getWarnings();
237 			if (warnings != null) {
238 				throw new SQLWarningException("Warning not ignored", warnings);
239 			}
240 		}
241 	}
242 
243 	/**
244 	 * Moves the cursor in the ResultSet to the position specified by the row
245 	 * parameter by traversing the ResultSet.
246 	 * @param row
247 	 */
248 	private void moveCursorToRow(int row) {
249 		try {
250 			int count = 0;
251 			while (row != count && rs.next()) {
252 				count++;
253 			}
254 		}
255 		catch (SQLException se) {
256 			throw getExceptionTranslator().translate("Attempted to move ResultSet to last committed row", getSql(), se);
257 		}
258 	}
259 
260 	/**
261 	 * Gives the JDBC driver a hint as to the number of rows that should be
262 	 * fetched from the database when more rows are needed for this
263 	 * <code>ResultSet</code> object. If the fetch size specified is zero, the
264 	 * JDBC driver ignores the value.
265 	 *
266 	 * @param fetchSize the number of rows to fetch
267 	 * @see ResultSet#setFetchSize(int)
268 	 */
269 	public void setFetchSize(int fetchSize) {
270 		this.fetchSize = fetchSize;
271 	}
272 
273 	/**
274 	 * Sets the limit for the maximum number of rows that any
275 	 * <code>ResultSet</code> object can contain to the given number.
276 	 *
277 	 * @param maxRows the new max rows limit; zero means there is no limit
278 	 * @see Statement#setMaxRows(int)
279 	 */
280 	public void setMaxRows(int maxRows) {
281 		this.maxRows = maxRows;
282 	}
283 
284 	/**
285 	 * Sets the number of seconds the driver will wait for a
286 	 * <code>Statement</code> object to execute to the given number of seconds.
287 	 * If the limit is exceeded, an <code>SQLException</code> is thrown.
288 	 *
289 	 * @param queryTimeout seconds the new query timeout limit in seconds; zero
290 	 * means there is no limit
291 	 * @see Statement#setQueryTimeout(int)
292 	 */
293 	public void setQueryTimeout(int queryTimeout) {
294 		this.queryTimeout = queryTimeout;
295 	}
296 
297 	/**
298 	 * Set whether SQLWarnings should be ignored (only logged) or exception
299 	 * should be thrown.
300 	 *
301 	 * @param ignoreWarnings if TRUE, warnings are ignored
302 	 */
303 	public void setIgnoreWarnings(boolean ignoreWarnings) {
304 		this.ignoreWarnings = ignoreWarnings;
305 	}
306 
307 	/**
308 	 * Allow verification of cursor position after current row is processed by
309 	 * RowMapper or RowCallbackHandler. Default value is TRUE.
310 	 *
311 	 * @param verifyCursorPosition if true, cursor position is verified
312 	 */
313 	public void setVerifyCursorPosition(boolean verifyCursorPosition) {
314 		this.verifyCursorPosition = verifyCursorPosition;
315 	}
316 
317 	/**
318 	 * Indicate whether the JDBC driver supports setting the absolute row on a
319 	 * {@link ResultSet}. It is recommended that this is set to
320 	 * <code>true</code> for JDBC drivers that supports ResultSet.absolute() as
321 	 * it may improve performance, especially if a step fails while working with
322 	 * a large data set.
323 	 *
324 	 * @see ResultSet#absolute(int)
325 	 *
326 	 * @param driverSupportsAbsolute <code>false</code> by default
327 	 */
328 	public void setDriverSupportsAbsolute(boolean driverSupportsAbsolute) {
329 		this.driverSupportsAbsolute = driverSupportsAbsolute;
330 	}
331 
332 	/**
333 	 * Indicate whether the connection used for the cursor should be used by all other processing
334 	 * thus sharing the same transaction. If this is set to false, which is the default, then the
335 	 * cursor will be opened using in its connection and will not participate in any transactions
336 	 * started for the rest of the step processing. If you set this flag to true then you must
337 	 * wrap the DataSource in a {@link ExtendedConnectionDataSourceProxy} to prevent the
338 	 * connection from being closed and released after each commit.
339 	 *
340 	 * When you set this option to <code>true</code> then the statement used to open the cursor
341 	 * will be created with both 'READ_ONLY' and 'HOLD_CUSORS_OVER_COMMIT' options. This allows
342 	 * holding the cursor open over transaction start and commits performed in the step processing.
343 	 * To use this feature you need a database that supports this and a JDBC driver supporting
344 	 * JDBC 3.0 or later.
345 	 *
346 	 * @param useSharedExtendedConnection <code>false</code> by default
347 	 */
348 	public void setUseSharedExtendedConnection(boolean useSharedExtendedConnection) {
349 		this.useSharedExtendedConnection = useSharedExtendedConnection;
350 	}
351 
352 	public boolean isUseSharedExtendedConnection() {
353 		return useSharedExtendedConnection;
354 	}
355 
356 	public abstract String getSql();
357 
358 	/**
359 	 * Check the result set is in synch with the currentRow attribute. This is
360 	 * important to ensure that the user hasn't modified the current row.
361 	 */
362 	private void verifyCursorPosition(long expectedCurrentRow) throws SQLException {
363 		if (verifyCursorPosition) {
364 			if (expectedCurrentRow != this.rs.getRow()) {
365 				throw new InvalidDataAccessResourceUsageException("Unexpected cursor position change.");
366 			}
367 		}
368 	}
369 
370 	/**
371 	 * Close the cursor and database connection. Make call to cleanupOnClose so sub classes can cleanup
372 	 * any resources they have allocated.
373 	 */
374 	@Override
375 	protected void doClose() throws Exception {
376 		initialized = false;
377 		JdbcUtils.closeResultSet(this.rs);
378 		rs = null;
379 		cleanupOnClose();
380 		if (useSharedExtendedConnection && dataSource instanceof ExtendedConnectionDataSourceProxy) {
381 			((ExtendedConnectionDataSourceProxy)dataSource).stopCloseSuppression(this.con);
382 			if (!TransactionSynchronizationManager.isActualTransactionActive()) {
383 				DataSourceUtils.releaseConnection(con, dataSource);
384 			}
385 		}
386 		else {
387 			JdbcUtils.closeConnection(this.con);
388 		}
389 	}
390 
391 	protected abstract void cleanupOnClose()  throws Exception;
392 
393 	/**
394 	 * Execute the statement to open the cursor.
395 	 */
396 	@Override
397 	protected final void doOpen() throws Exception {
398 
399 		Assert.state(!initialized, "Stream is already initialized.  Close before re-opening.");
400 		Assert.isNull(rs, "ResultSet still open!  Close before re-opening.");
401 
402 		initializeConnection();
403 		openCursor(con);
404 		initialized = true;
405 
406 	}
407 
408 	protected void initializeConnection() {
409 		Assert.state(getDataSource() != null, "DataSource must not be null.");
410 
411 		try {
412 			if (useSharedExtendedConnection) {
413 				if (!(getDataSource() instanceof ExtendedConnectionDataSourceProxy)) {
414 					throw new InvalidDataAccessApiUsageException(
415 							"You must use a ExtendedConnectionDataSourceProxy for the dataSource when " +
416 							"useSharedExtendedConnection is set to true.");
417 				}
418 				this.con = DataSourceUtils.getConnection(dataSource);
419 				((ExtendedConnectionDataSourceProxy)dataSource).startCloseSuppression(this.con);
420 			}
421 			else {
422 				this.con = dataSource.getConnection();
423 			}
424 		}
425 		catch (SQLException se) {
426 			close();
427 			throw getExceptionTranslator().translate("Executing query", getSql(), se);
428 		}
429 	}
430 
431 	protected abstract void openCursor(Connection con);
432 
433 	/**
434 	 * Read next row and map it to item, verify cursor position if
435 	 * {@link #setVerifyCursorPosition(boolean)} is true.
436 	 */
437 	@Override
438 	protected T doRead() throws Exception {
439 		if (rs == null) {
440 			throw new ReaderNotOpenException("Reader must be open before it can be read.");
441 		}
442 
443 		try {
444 			if (!rs.next()) {
445 				return null;
446 			}
447 			int currentRow = getCurrentItemCount();
448 			T item = readCursor(rs, currentRow);
449 			verifyCursorPosition(currentRow);
450 			return item;
451 		}
452 		catch (SQLException se) {
453 			throw getExceptionTranslator().translate("Attempt to process next row failed", getSql(), se);
454 		}
455 	}
456 
457 	/**
458 	 * Read the cursor and map to the type of object this reader should return. This method must be
459 	 * overriden by subclasses.
460 	 *
461 	 * @param rs The current result set
462 	 * @param currentRow Current position of the result set
463 	 * @return the mapped object at the cursor position
464 	 * @throws SQLException
465 	 */
466 	protected abstract T readCursor(ResultSet rs, int currentRow) throws SQLException;
467 
468 	/**
469 	 * Use {@link ResultSet#absolute(int)} if possible, otherwise scroll by
470 	 * calling {@link ResultSet#next()}.
471 	 */
472 	@Override
473 	protected void jumpToItem(int itemIndex) throws Exception {
474 		if (driverSupportsAbsolute) {
475 			try {
476 				rs.absolute(itemIndex);
477 			}
478 			catch (SQLException e) {
479 				// Driver does not support rs.absolute(int) revert to
480 				// traversing ResultSet
481 				log.warn("The JDBC driver does not appear to support ResultSet.absolute(). Consider"
482 						+ " reverting to the default behavior setting the driverSupportsAbsolute to false", e);
483 
484 				moveCursorToRow(itemIndex);
485 			}
486 		}
487 		else {
488 			moveCursorToRow(itemIndex);
489 		}
490 	}
491 
492 }