eskabetxe commented on code in PR #209: URL: https://github.com/apache/flink-connector-jdbc/pull/209#discussion_r3785366694
########## flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/connection/AbstractConnectionProvider.java: ########## @@ -0,0 +1,447 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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.apache.flink.connector.jdbc.core.datastream.connection; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.snapshot.domain.TableColumn; +import org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.snapshot.domain.TableId; +import org.apache.flink.util.Preconditions; + +import com.zaxxer.hikari.HikariConfig; +import com.zaxxer.hikari.HikariDataSource; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nonnull; + +import java.sql.Connection; +import java.sql.Driver; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.sql.Statement; +import java.time.Duration; +import java.util.Enumeration; +import java.util.HashSet; +import java.util.LinkedHashSet; +import java.util.Map; +import java.util.Objects; +import java.util.Properties; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +/** + * Base {@link ConnectionProvider} that implements the connection lifecycle, pooling, and + * prepared-statement plumbing shared by every dialect-specific provider. Subclasses only need to + * implement the dialect-specific table discovery and bound-query building methods declared by + * {@link ConnectionProvider} (e.g. {@code getTables}, {@code queryMinMax}, {@code + * queryNextChunkMax}, {@code createQueryWithBounds}, {@code newInstance}). + */ +@Internal +public abstract class AbstractConnectionProvider implements ConnectionProvider { + + private static final Logger LOG = LoggerFactory.getLogger(AbstractConnectionProvider.class); + + private static final int DEFAULT_POOL_SIZE = 4; + private static final int MINIMUM_POOL_SIZE = 1; + + protected final ConnectionOptions jdbcOptions; + private int queryTimeoutSeconds; + private transient Driver loadedDriver; + private transient Connection connection; + private final Map<String, PreparedStatement> statementCache; + private transient HikariDataSource connectionPool; + private final boolean poolOwner; + + static { + // Load DriverManager first to avoid deadlock between DriverManager's + // static initialization block and specific driver class's static + // initialization block when two different driver classes are loading + // concurrently using Class.forName while DriverManager is uninitialized + // before. + // + // This could happen in JDK 8 but not above as driver loading has been + // moved out of DriverManager's static initialization block since JDK 9. + DriverManager.getDrivers(); + } + + protected AbstractConnectionProvider(ConnectionOptions jdbcOptions) { + this.jdbcOptions = jdbcOptions; + this.statementCache = new ConcurrentHashMap<>(); + this.queryTimeoutSeconds = jdbcOptions.getConnectionQueryTimeoutSeconds(); + this.poolOwner = true; + try { + // Establish the raw connection directly rather than via getOrEstablishConnection(): + // that method invokes the overridable onConnectionEstablished() hook, which a + // subclass may depend on its own fields for — fields that aren't initialized yet + // while this superclass constructor is still running. + establishConnection(); + } catch (Exception e) { + throw new ConnectionException( + "Failed to establish initial connection during connection provider construction.", + e); + } + } + + /** + * Creates a pooled instance that borrows its connection from the given pool. This instance does + * NOT own the pool and will not shut it down on close. + */ + protected AbstractConnectionProvider(ConnectionOptions jdbcOptions, HikariDataSource pool) { + this.jdbcOptions = jdbcOptions; + this.statementCache = new ConcurrentHashMap<>(); + this.queryTimeoutSeconds = jdbcOptions.getConnectionQueryTimeoutSeconds(); + this.connectionPool = pool; + this.poolOwner = false; + try { + this.connection = pool.getConnection(); + } catch (SQLException e) { + throw new ConnectionException("Failed to borrow connection from pool.", e); + } + } + + protected synchronized HikariDataSource getOrCreatePool() { + if (connectionPool == null) { + connectionPool = createConnectionPool(); + } + return connectionPool; + } + + private HikariDataSource createConnectionPool() { + HikariConfig config = new HikariConfig(); + // Provide a DataSource directly so HikariCP doesn't try to load the driver + // via its own classloader. In Flink, the JDBC driver lives in the user + // classloader and is not visible to HikariCP's threads. Using this driver + // delegate guarantees connections are created with the same code path as + // getOrEstablishConnection(). + config.setDataSource(new ConnectionDataSource(jdbcOptions, this::getLoadedDriver)); + config.setMinimumIdle(MINIMUM_POOL_SIZE); + config.setMaximumPoolSize(maxPoolSize()); + config.setConnectionTimeout( + Duration.ofSeconds(jdbcOptions.getConnectionCheckTimeoutSeconds()).toMillis()); + config.setPoolName(poolName()); + LOG.info("Creating HikariCP connection pool with maxPoolSize={}", maxPoolSize()); + return new HikariDataSource(config); + } + + /** Maximum number of pooled connections. Override to change the pool size. */ + protected int maxPoolSize() { + return DEFAULT_POOL_SIZE; + } + + /** Name used for the HikariCP pool, shown in logs/metrics. */ + protected String poolName() { + return getClass().getSimpleName() + "-pool"; + } + + /** + * Hook invoked every time a connection is (re-)established, before it's handed back to the + * caller. No-op by default; dialect-specific subclasses can override to re-apply + * connection-scoped state (e.g. re-syncing a shared snapshot transaction id). + */ + protected void onConnectionEstablished() throws SQLException {} + + @Override + public Connection getConnection() { + return connection; + } + + @Nonnull + @Override + public Properties getProperties() { + return jdbcOptions.getProperties(); + } + + @Override + public boolean isConnectionValid() throws SQLException { + return connection != null + && !connection.isClosed() + && connection.isValid(jdbcOptions.getConnectionCheckTimeoutSeconds()); + } + + private Driver loadDriver(String driverName) throws SQLException, ClassNotFoundException { + Preconditions.checkNotNull(driverName); + Enumeration<Driver> drivers = DriverManager.getDrivers(); + while (drivers.hasMoreElements()) { + Driver driver = drivers.nextElement(); + if (driver.getClass().getName().equals(driverName)) { + return driver; + } + } + // We could reach here for reasons: + // * Class loader hell of DriverManager(see JDK-8146872). + // * driver is not installed as a service provider. + Class<?> clazz = + Class.forName(driverName, true, Thread.currentThread().getContextClassLoader()); + try { + return (Driver) clazz.getDeclaredConstructor().newInstance(); + } catch (Exception ex) { + throw new SQLException("Fail to create driver of class " + driverName, ex); + } + } + + private Driver getLoadedDriver() throws SQLException, ClassNotFoundException { + if (loadedDriver == null) { + loadedDriver = loadDriver(jdbcOptions.getDriverName()); + } + return loadedDriver; + } + + @Override + public Connection getOrEstablishConnection() throws SQLException, ClassNotFoundException { + if (isConnectionValid()) { + onConnectionEstablished(); + return connection; + } + establishConnection(); + onConnectionEstablished(); + return connection; + } + + /** + * Establishes a fresh {@link #connection}. Does not invoke {@link #onConnectionEstablished()}. + */ + private Connection establishConnection() throws SQLException, ClassNotFoundException { Review Comment: yes, can be void.. changed -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
