[Improve]Release JDBC connection resources

apache · Aug 21, 2024 · 35c5637 · 35c5637
1 parent 4d9b45a
commit 35c5637
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 15 deletions.
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/doris/DorisSystem.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/doris/DorisSystem.java
@@ -30,6 +30,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.Serializable;
+import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.Statement;
@@ -103,10 +104,11 @@ public void createTable(TableSchema schema) {
     }
 
     public void execute(String sql) {
-        try (Statement statement =
-                jdbcConnectionProvider.getOrEstablishConnection().createStatement()) {
+        try (Connection connection = jdbcConnectionProvider.getOrEstablishConnection();
+                Statement statement = connection.createStatement()) {
             statement.execute(sql);
         } catch (Exception e) {
+            LOG.error("SQL query could not be executed: {}", sql, e);
             throw new DorisSystemException(
                     String.format("SQL query could not be executed: %s", sql), e);
         }
@@ -116,18 +118,19 @@ public List<String> extractColumnValuesBySQL(
             String sql, int columnIndex, Predicate<String> filterFunc, Object... params) {
 
         List<String> columnValues = Lists.newArrayList();
-        try (PreparedStatement ps =
-                jdbcConnectionProvider.getOrEstablishConnection().prepareStatement(sql)) {
+        try (Connection connection = jdbcConnectionProvider.getOrEstablishConnection();
+                PreparedStatement ps = connection.prepareStatement(sql)) {
             if (Objects.nonNull(params) && params.length > 0) {
                 for (int i = 0; i < params.length; i++) {
                     ps.setObject(i + 1, params[i]);
                 }
             }
-            ResultSet rs = ps.executeQuery();
-            while (rs.next()) {
-                String columnValue = rs.getString(columnIndex);
-                if (Objects.isNull(filterFunc) || filterFunc.test(columnValue)) {
-                    columnValues.add(columnValue);
+            try (ResultSet rs = ps.executeQuery()) {
+                while (rs.next()) {
+                    String columnValue = rs.getString(columnIndex);
+                    if (filterFunc == null || filterFunc.test(columnValue)) {
+                        columnValues.add(columnValue);
+                    }
                 }
             }
             return columnValues;
@@ -152,16 +155,17 @@ public Map<String, String> getTableFieldNames(String databaseName, String tableN
                         databaseName, tableName);
 
         Map<String, String> columnValues = new HashMap<>();
-        try (PreparedStatement ps =
-                jdbcConnectionProvider.getOrEstablishConnection().prepareStatement(sql)) {
-            ResultSet rs = ps.executeQuery();
+        try (Connection connection = jdbcConnectionProvider.getOrEstablishConnection();
+                PreparedStatement ps = connection.prepareStatement(sql);
+                ResultSet rs = ps.executeQuery()) {
             while (rs.next()) {
-                String filedName = rs.getString(1);
+                String fieldName = rs.getString(1);
                 String datatype = rs.getString(2);
-                columnValues.put(filedName, datatype);
+                columnValues.put(fieldName, datatype);
             }
             return columnValues;
         } catch (Exception e) {
+            LOG.error("SQL query could not be executed: {}", sql, e);
             throw new DorisSystemException(
                     String.format("The following SQL query could not be executed: %s", sql), e);
         }

diff --git a/...nnector/src/main/java/org/apache/doris/flink/connection/SimpleJdbcConnectionProvider.java b/...nnector/src/main/java/org/apache/doris/flink/connection/SimpleJdbcConnectionProvider.java
@@ -42,7 +42,7 @@ public SimpleJdbcConnectionProvider(DorisConnectionOptions options) {
 
     @Override
     public Connection getOrEstablishConnection() throws ClassNotFoundException, SQLException {
-        if (connection != null && !connection.isClosed() && connection.isValid(10000)) {
+        if (connection != null && !connection.isClosed() && connection.isValid(10)) {
             return connection;
         }
         try {