BATCH-970: Added an ExtendedConnectionDataSourceProxy to be used with the JdbcCursorItemReader to allow the cursor and any other processing to share transactions.
This commit is contained in:
trisberg
@@ -0,0 +1,271 @@
|
||||
/*
|
||||
* Copyright 2002-2008 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.io.PrintWriter;
|
||||
import java.lang.reflect.InvocationHandler;
|
||||
import java.lang.reflect.InvocationTargetException;
|
||||
import java.lang.reflect.Method;
|
||||
import java.lang.reflect.Proxy;
|
||||
import java.sql.Connection;
|
||||
import java.sql.SQLException;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
|
||||
import org.springframework.jdbc.datasource.ConnectionProxy;
|
||||
import org.springframework.jdbc.datasource.DataSourceUtils;
|
||||
import org.springframework.jdbc.datasource.SmartDataSource;
|
||||
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
||||
|
||||
/**
|
||||
* Implementation of {@link SmartDataSource} that is capable of keeping a single JDBC Connection
|
||||
* which is NOT closed after each use even if {@link Connection#close()} is called.
|
||||
*
|
||||
* The connection can be kept open over multiple transactions when used together with any of Spring's
|
||||
* {@link org.springframework.transaction.PlatformTransactionManager} implementations.
|
||||
*
|
||||
* <p>Loosely based on the SingleConnectionDataSource implementation in Spring Core. Intended
|
||||
* to be used with the {@link JdbcCursorItemReader} to provide a connection that remains
|
||||
* open across transaction boundaries, It remains open for the life of the cursor, and can be
|
||||
* shared with the main transaction of the rest of the step processing.
|
||||
*
|
||||
* <p>Once close suppression has been turned on for a connection, it will be returned for the first
|
||||
* {@link #getConnection()} call. Any subsequent calls to {@link #getConnection()} will retrieve a
|
||||
* new connection from the wrapped {@link DataSource} until the {@link DataSourceUtils} queries
|
||||
* whether the connection should be closed or not by calling {@link #shouldClose(Connection)} for
|
||||
* the close-suppressed {@link Connection}. At that point the cycle starts over again, and the next
|
||||
* {@link #getConnection()} call will have the {@link Connection} that is being close-suppressed
|
||||
* returned. This allows the use of the close-suppressed {@link Connection} to be the main
|
||||
* {@link Connection} for an extended data access process. The close suppression is turned off by
|
||||
* calling {@link #stopCloseSuppression(Connection)}.
|
||||
*
|
||||
* <p>This class is not multi-threading capable.
|
||||
*
|
||||
* <p>The connection returned will be a close-suppressing proxy instead of the physical
|
||||
* {@link Connection}. Be aware that you will not be able to cast this to a native
|
||||
* <code>OracleConnection</code> or the like anymore; you need to use a
|
||||
* {@link org.springframework.jdbc.support.nativejdbc.NativeJdbcExtractor}.
|
||||
*
|
||||
* @author Thomas Risberg
|
||||
* @see #getConnection()
|
||||
* @see java.sql.Connection#close()
|
||||
* @see DataSourceUtils#releaseConnection
|
||||
* @see org.springframework.jdbc.support.nativejdbc.NativeJdbcExtractor
|
||||
*/
|
||||
public class ExtendedConnectionDataSourceProxy implements SmartDataSource {
|
||||
|
||||
/** Provided DataSource */
|
||||
private DataSource dataSource;
|
||||
|
||||
/** The connection to suppress close calls for */
|
||||
private Connection closeSuppressedConnection = null;
|
||||
|
||||
/** The connection to suppress close calls for */
|
||||
private boolean borrowedConnection = false;
|
||||
|
||||
/** Synchronization monitor for the shared Connection */
|
||||
private final Object connectionMonitor = new Object();
|
||||
|
||||
/**
|
||||
* No arg constructor for use when configured using JavaBean style.
|
||||
*/
|
||||
public ExtendedConnectionDataSourceProxy() {
|
||||
}
|
||||
|
||||
/**
|
||||
* Constructor that takes as a parameter with the {&link DataSource} to be wrapped.
|
||||
*/
|
||||
public ExtendedConnectionDataSourceProxy(DataSource dataSource) {
|
||||
this.dataSource = dataSource;
|
||||
}
|
||||
|
||||
/**
|
||||
* Setter for the {&link DataSource} that is to be wrapped.
|
||||
*
|
||||
* @param dataSource the DataSource
|
||||
*/
|
||||
public void setDataSource(DataSource dataSource) {
|
||||
this.dataSource = dataSource;
|
||||
}
|
||||
|
||||
/**
|
||||
* @see SmartDataSource
|
||||
*/
|
||||
public boolean shouldClose(Connection connection) {
|
||||
boolean shouldClose = !isCloseSuppressionActive(connection);
|
||||
if (borrowedConnection && closeSuppressedConnection.equals(connection)) {
|
||||
borrowedConnection = false;
|
||||
}
|
||||
return shouldClose;
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the status of close suppression being activated for a given {@link Connection}
|
||||
*
|
||||
* @param connection the {@link Connection} that the close suppression status is requested for
|
||||
* @return true or false
|
||||
*/
|
||||
public boolean isCloseSuppressionActive(Connection connection) {
|
||||
if (connection == null) {
|
||||
return false;
|
||||
}
|
||||
return connection.equals(closeSuppressedConnection) ? true : false;
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @param connection the {@link Connection} that close suppression is requested for
|
||||
*/
|
||||
public void startCloseSuppression(Connection connection) {
|
||||
synchronized (this.connectionMonitor) {
|
||||
closeSuppressedConnection = connection;
|
||||
if(TransactionSynchronizationManager.isActualTransactionActive()) {
|
||||
borrowedConnection = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @param connection the {@link Connection} that close suppression should be turned off for
|
||||
*/
|
||||
public void stopCloseSuppression(Connection connection) {
|
||||
synchronized (this.connectionMonitor) {
|
||||
closeSuppressedConnection = null;
|
||||
borrowedConnection = false;
|
||||
}
|
||||
}
|
||||
|
||||
public Connection getConnection() throws SQLException {
|
||||
synchronized (this.connectionMonitor) {
|
||||
return initConnection(null, null);
|
||||
}
|
||||
}
|
||||
|
||||
public Connection getConnection(String username, String password) throws SQLException {
|
||||
synchronized (this.connectionMonitor) {
|
||||
return initConnection(username, password);
|
||||
}
|
||||
}
|
||||
|
||||
private boolean completeCloseCall(Connection connection) {
|
||||
if (borrowedConnection && closeSuppressedConnection.equals(connection)) {
|
||||
borrowedConnection = false;
|
||||
}
|
||||
return isCloseSuppressionActive(connection);
|
||||
}
|
||||
|
||||
private Connection initConnection(String username, String password) throws SQLException {
|
||||
if (closeSuppressedConnection != null) {
|
||||
if (!borrowedConnection) {
|
||||
borrowedConnection = true;
|
||||
return closeSuppressedConnection;
|
||||
}
|
||||
}
|
||||
Connection target;
|
||||
if (username != null) {
|
||||
target = dataSource.getConnection(username, password);
|
||||
}
|
||||
else {
|
||||
target = dataSource.getConnection();
|
||||
}
|
||||
Connection connection = getCloseSuppressingConnectionProxy(target);
|
||||
return connection;
|
||||
}
|
||||
|
||||
public PrintWriter getLogWriter() throws SQLException {
|
||||
return dataSource.getLogWriter();
|
||||
}
|
||||
|
||||
public int getLoginTimeout() throws SQLException {
|
||||
return dataSource.getLoginTimeout();
|
||||
}
|
||||
|
||||
public void setLogWriter(PrintWriter out) throws SQLException {
|
||||
dataSource.setLogWriter(out);
|
||||
}
|
||||
|
||||
public void setLoginTimeout(int seconds) throws SQLException {
|
||||
dataSource.setLoginTimeout(seconds);
|
||||
}
|
||||
|
||||
/**
|
||||
* Wrap the given Connection with a proxy that delegates every method call to it
|
||||
* but suppresses close calls.
|
||||
* @param target the original Connection to wrap
|
||||
* @return the wrapped Connection
|
||||
*/
|
||||
protected Connection getCloseSuppressingConnectionProxy(Connection target) {
|
||||
return (Connection) Proxy.newProxyInstance(
|
||||
ConnectionProxy.class.getClassLoader(),
|
||||
new Class[] {ConnectionProxy.class},
|
||||
new CloseSuppressingInvocationHandler(target, this));
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Invocation handler that suppresses close calls on JDBC Connections until the associated instance of the
|
||||
* ExtendedConnectionDataSourceProxy determines the connection should actually be closed.
|
||||
*/
|
||||
private static class CloseSuppressingInvocationHandler implements InvocationHandler {
|
||||
|
||||
private final Connection target;
|
||||
|
||||
private final ExtendedConnectionDataSourceProxy dataSource;
|
||||
|
||||
public CloseSuppressingInvocationHandler(Connection target, ExtendedConnectionDataSourceProxy dataSource) {
|
||||
this.dataSource = dataSource;
|
||||
this.target = target;
|
||||
}
|
||||
|
||||
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
|
||||
// Invocation on ConnectionProxy interface coming in...
|
||||
|
||||
if (method.getName().equals("equals")) {
|
||||
// Only consider equal when proxies are identical.
|
||||
return (proxy == args[0] ? Boolean.TRUE : Boolean.FALSE);
|
||||
}
|
||||
else if (method.getName().equals("hashCode")) {
|
||||
// Use hashCode of Connection proxy.
|
||||
return new Integer(System.identityHashCode(proxy));
|
||||
}
|
||||
else if (method.getName().equals("close")) {
|
||||
// Handle close method: don't pass the call on if we are suppressing close calls.
|
||||
if (dataSource.completeCloseCall((Connection)proxy)) {
|
||||
return null;
|
||||
}
|
||||
else {
|
||||
target.close();
|
||||
return null;
|
||||
}
|
||||
}
|
||||
else if (method.getName().equals("getTargetConnection")) {
|
||||
// Handle getTargetConnection method: return underlying Connection.
|
||||
return this.target;
|
||||
}
|
||||
|
||||
// Invoke method on target Connection.
|
||||
try {
|
||||
return method.invoke(this.target, args);
|
||||
}
|
||||
catch (InvocationTargetException ex) {
|
||||
throw ex.getTargetException();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -31,6 +31,7 @@ import org.springframework.batch.item.ExecutionContext;
|
||||
import org.springframework.batch.item.ItemStream;
|
||||
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;
|
||||
@@ -40,15 +41,38 @@ 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
|
||||
* 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.
|
||||
* 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
|
||||
* in any transactions created as part of the step processing.
|
||||
* </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>
|
||||
@@ -56,12 +80,10 @@ import org.springframework.util.ClassUtils;
|
||||
* 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
|
||||
* 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 line, regardless of rollbacks or restarts.
|
||||
* ResultSet at the correct row, regardless of rollbacks or restarts.
|
||||
* </p>
|
||||
*
|
||||
* <p>
|
||||
@@ -73,30 +95,22 @@ import org.springframework.util.ClassUtils;
|
||||
* </p>
|
||||
*
|
||||
* <p>
|
||||
* Transactions: The same ResultSet is held open regardless of commits or roll
|
||||
* backs in a surrounding transaction. When a transaction is committed, the
|
||||
* reader will be notified through the {@link #update(ExecutionContext)} so that
|
||||
* it can save it's current row number. Clients of this reader are responsible
|
||||
* for buffering the items in the case that they need to be re-presented on a
|
||||
* rollback.
|
||||
* </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.
|
||||
* 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
|
||||
* throws an exception.
|
||||
* is not available for 'TYPE_FORWARD_ONLY' result sets.
|
||||
* </p>
|
||||
*
|
||||
* @author Lucas Ward
|
||||
* @author Peter Zozom
|
||||
* @author Robert Kasanicky
|
||||
* @author Thomas Risberg
|
||||
*/
|
||||
public class JdbcCursorItemReader<T> extends AbstractItemCountingItemStreamItemReader<T> implements InitializingBean {
|
||||
|
||||
@@ -134,7 +148,7 @@ public class JdbcCursorItemReader<T> extends AbstractItemCountingItemStreamItemR
|
||||
|
||||
private boolean driverSupportsAbsolute = false;
|
||||
|
||||
private boolean participateInExistingTransaction = false;
|
||||
private boolean useSharedExtendedConnection = false;
|
||||
|
||||
public JdbcCursorItemReader() {
|
||||
setName(ClassUtils.getShortName(JdbcCursorItemReader.class));
|
||||
@@ -162,25 +176,28 @@ public class JdbcCursorItemReader<T> extends AbstractItemCountingItemStreamItemR
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
private void executeQuery() {
|
||||
|
||||
Assert.state(dataSource != null, "DataSource must not be null.");
|
||||
|
||||
try {
|
||||
if (participateInExistingTransaction) {
|
||||
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);
|
||||
}
|
||||
preparedStatement = this.con.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY,
|
||||
ResultSet.HOLD_CURSORS_OVER_COMMIT);
|
||||
applyStatementSettings(preparedStatement);
|
||||
if (this.preparedStatementSetter != null) {
|
||||
preparedStatementSetter.setValues(preparedStatement);
|
||||
@@ -385,16 +402,23 @@ public class JdbcCursorItemReader<T> extends AbstractItemCountingItemStreamItemR
|
||||
}
|
||||
|
||||
/**
|
||||
* Indicate whether the cursor should be opened as part of an existing transaction or if it
|
||||
* should be opened in its own transaction. The default is for the cursor to be opened in its
|
||||
* own transaction. If you set this flag to true then you should wrap the DataSource in a
|
||||
* {@link org.springframework.jdbc.datasource.SingleConnectionDataSource} to prevent the
|
||||
* connection from being closed after each commit.
|
||||
* 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 participateInExistingTransaction <code>false</code> by default
|
||||
* @param useSharedExtendedConnection <code>false</code> by default
|
||||
*/
|
||||
public void setParticipateInExistingTransaction(boolean participateInExistingTransaction) {
|
||||
this.participateInExistingTransaction = participateInExistingTransaction;
|
||||
public void setUseSharedExtendedConnection(boolean useSharedExtendedConnection) {
|
||||
this.useSharedExtendedConnection = useSharedExtendedConnection;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -415,10 +439,17 @@ public class JdbcCursorItemReader<T> extends AbstractItemCountingItemStreamItemR
|
||||
protected void doClose() throws Exception {
|
||||
initialized = false;
|
||||
JdbcUtils.closeResultSet(this.rs);
|
||||
JdbcUtils.closeStatement(this.preparedStatement);
|
||||
JdbcUtils.closeConnection(this.con);
|
||||
rs = null;
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user