This is an automated email from the ASF dual-hosted git repository. jark pushed a commit to branch release-1.12 in repository https://gitbox.apache.org/repos/asf/flink.git
commit 39b53f672d7fa745c331012643a8de28b02b2821 Author: Kezhu Wang <[email protected]> AuthorDate: Thu Dec 10 21:05:33 2020 +0800 [FLINK-19435][connectors/jdbc] Fix deadlock when loading different sql driver classes concurrently using Class.forName This closes #14361 --- .../org/apache/flink/connector/jdbc/JdbcInputFormat.java | 6 ++++++ .../internal/connection/SimpleJdbcConnectionProvider.java | 12 ++++++++++++ .../flink/connector/jdbc/table/JdbcLookupFunction.java | 6 ++++++ .../connector/jdbc/table/JdbcRowDataLookupFunction.java | 6 ++++++ 4 files changed, 30 insertions(+) diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcInputFormat.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcInputFormat.java index 7665600..0982a84 100644 --- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcInputFormat.java +++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcInputFormat.java @@ -141,6 +141,12 @@ public class JdbcInputFormat extends RichInputFormat<Row, InputSplit> implements public void openInputFormat() { //called once per inputFormat (on open) try { + // Load DriverManager first to avoid deadlock between DriverManager's + // static initialization block and specific driver class's static + // initialization block. + // + // See comments in SimpleJdbcConnectionProvider for more details. + DriverManager.getDrivers(); Class.forName(drivername); if (username == null) { dbConn = DriverManager.getConnection(dbURL); diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/SimpleJdbcConnectionProvider.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/SimpleJdbcConnectionProvider.java index d9bac06..4424054 100644 --- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/SimpleJdbcConnectionProvider.java +++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/SimpleJdbcConnectionProvider.java @@ -40,6 +40,18 @@ public class SimpleJdbcConnectionProvider implements JdbcConnectionProvider, Ser private transient volatile Connection connection; + 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(); + } + public SimpleJdbcConnectionProvider(JdbcConnectionOptions jdbcOptions) { this.jdbcOptions = jdbcOptions; } diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcLookupFunction.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcLookupFunction.java index 422ca21..037003b 100644 --- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcLookupFunction.java +++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcLookupFunction.java @@ -205,6 +205,12 @@ public class JdbcLookupFunction extends TableFunction<Row> { } private void establishConnectionAndStatement() throws SQLException, ClassNotFoundException { + // Load DriverManager first to avoid deadlock between DriverManager's + // static initialization block and specific driver class's static + // initialization block. + // + // See comments in SimpleJdbcConnectionProvider for more details. + DriverManager.getDrivers(); Class.forName(drivername); if (username == null) { dbConn = DriverManager.getConnection(dbURL); diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunction.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunction.java index 50bc550..d00bb47 100644 --- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunction.java +++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunction.java @@ -195,6 +195,12 @@ public class JdbcRowDataLookupFunction extends TableFunction<RowData> { } private void establishConnectionAndStatement() throws SQLException, ClassNotFoundException { + // Load DriverManager first to avoid deadlock between DriverManager's + // static initialization block and specific driver class's static + // initialization block. + // + // See comments in SimpleJdbcConnectionProvider for more details. + DriverManager.getDrivers(); Class.forName(drivername); if (username == null) { dbConn = DriverManager.getConnection(dbURL);
