och5351 commented on code in PR #209: URL: https://github.com/apache/flink-connector-jdbc/pull/209#discussion_r3783836316
########## flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/snapshot/AsyncSnapshotSplitterEnumerator.java: ########## @@ -0,0 +1,189 @@ +/* + * 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.source.enumerator.splitter.snapshot; + +import org.apache.flink.api.connector.source.Boundedness; +import org.apache.flink.connector.jdbc.core.datastream.connection.ConnectionProvider; +import org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.SplitterEnumerator; +import org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit; +import org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Shared async background-computation machinery for the snapshot splitters ({@link + * TableSplitterEnumerator}, {@link DatabaseSplitterEnumerator}): a single background daemon thread + * computes {@code T} items and offers them into a queue, while {@link #enumerateSplits()} drains + * and converts whatever's ready — non-blocking apart from a short wait for the very first item. + * + * @param <T> the type of item the background thread produces, converted to a {@link + * JdbcSourceSplit} at drain time via {@link #toSplit} + */ +abstract class AsyncSnapshotSplitterEnumerator<T> implements SplitterEnumerator { + + private final Logger log = LoggerFactory.getLogger(getClass()); + private final String name; + private final Queue<T> outputQueue = new ConcurrentLinkedQueue<>(); + + protected transient ConnectionProvider connection; + + private transient ExecutorService executor; + private transient AtomicBoolean workDone; + private transient CountDownLatch firstReady; + private transient volatile Throwable backgroundFailure; + + protected AsyncSnapshotSplitterEnumerator(String name) { + this.name = name; + } + + /** Validates and stores the connection provider for use by subclasses. */ + protected final void initConnection(JdbcConnectionProvider connectionProvider) { + if (!(connectionProvider instanceof ConnectionProvider)) { + throw new IllegalArgumentException( + "Connection provider must be an instance of " + + ConnectionProvider.class.getSimpleName()); + } + this.connection = (ConnectionProvider) connectionProvider; + } + + @Override + public final Boundedness getBoundedness() { + return Boundedness.CONTINUOUS_UNBOUNDED; Review Comment: Snapshot reads are finite, so this should return BOUNDED. With CONTINUOUS_UNBOUNDED the job never terminates after all chunks are emitted. ########## flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/snapshot/TableSplitterEnumerator.java: ########## @@ -0,0 +1,212 @@ +/* + * 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.source.enumerator.splitter.snapshot; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.SplitterEnumerator; +import org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.snapshot.domain.TableBounds; +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.connector.jdbc.core.datastream.source.split.CheckpointedOffset; +import org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit; +import org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * An implementation of {@link SplitterEnumerator} that splits a table into multiple splits based on + * the primary key column and a specified chunk size. + */ +@PublicEvolving +public class TableSplitterEnumerator extends AsyncSnapshotSplitterEnumerator<TableBounds> { + private final Logger log = LoggerFactory.getLogger(TableSplitterEnumerator.class); + + private final TableId tableId; + private final Set<String> columnNames; + private final int chunkSize; + + private TableColumn tablePrimaryKey; + private final Set<String> lineageQueries; + + private TableBounds tableMinMax; + private Object currentLowerBound; + private boolean boundsInitialized; + + protected TableSplitterEnumerator(TableId tableId, Set<String> columnNames, int chunkSize) { Review Comment: DatabaseSplitterEnumerator.prepareTableSplitters() already calls connection.getTableColumns() but strips it to Set<String> before passing to the constructor. Then TableSplitterEnumerator.validateTableAndColumns() calls getTableColumns() again for the same table. How about adding a second constructor that accepts pre-loaded Set<TableColumn>? ```suggestion // standalone usage — fetches lazily as before protected TableSplitterEnumerator(TableId tableId, Set<String> columnNames, int chunkSize) { ... this.discoveredColumns = null; } // called from DatabaseSplitterEnumerator with pre-loaded columns protected TableSplitterEnumerator(TableId tableId, Set<TableColumn> preloadedColumns, int chunkSize) { ... this.discoveredColumns = preloadedColumns; } private void validateTableAndColumns() { if (this.discoveredColumns == null) { this.discoveredColumns = connection.getTableColumns(tableId); } ... } ``` This avoids the redundant metadata query when called from DatabaseSplitterEnumerator. ########## 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 { + String connectionUrl = jdbcOptions.getDbURL(); + if (jdbcOptions.getDriverName() == null) { + connection = DriverManager.getConnection(connectionUrl, getProperties()); + } else { + Driver driver = getLoadedDriver(); + connection = driver.connect(connectionUrl, getProperties()); + if (connection == null) { + // Throw same exception as DriverManager.getConnection when no driver found to match + // caller expectation. + throw new SQLException("No suitable driver found for " + connectionUrl, "08001"); + } + } + return connection; + } + + @Override + public Connection reestablishConnection() throws SQLException, ClassNotFoundException { + closeConnection(); Review Comment: Edge case, but worth noting: on a 10s+ network outage (e.g., RDS failover), closeConnection() times out, aborts, and nulls this.connection. reestablishConnection() then falls through to establishConnection() -> DriverManager.getConnection(), bypassing the HikariCP pool entirely. For pool-borrowed instances, wouldn't connectionPool.getConnection() be more appropriate here? ########## 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: It's maybe void. isn't it? -- 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]
