[fix][worker]fix statement is closed before resultSet.getMetaData()

huaangcg · huaangcg · commit 2c88444e6945 · 2022-05-12T15:34:17.000+08:00
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java
@@ -180,6 +180,7 @@ public void executeFuncAndSql(List<SqlBinds> mainStatementsBinds,
                                   List<String> createFuncs) throws Exception {
         Connection connection = null;
         ResultSet resultSet = null;
+        PreparedStatement statement = null;
         try {
 
             // create connection
@@ -196,8 +197,10 @@ public void executeFuncAndSql(List<SqlBinds> mainStatementsBinds,
             String result = null;
             // decide whether to executeQuery or executeUpdate based on sqlType
             if (sqlParameters.getSqlType() == SqlType.QUERY.ordinal()) {
-                // query statements need to be convert to JsonArray and inserted into Alert to send
-                resultSet = executeQuery(connection, mainStatementsBinds.get(0), "main");
+                // query statements need to be converted to JsonArray and inserted into Alert to send
+                // statement must close after resultProcess
+                statement = prepareStatementAndBind(connection, mainStatementsBinds.get(0));
+                resultSet = executeQuery(mainStatementsBinds.get(0), "main", statement);
                 result = resultProcess(resultSet);
             } else if (sqlParameters.getSqlType() == SqlType.NON_QUERY.ordinal()) {
                 // non query statement
@@ -213,7 +216,7 @@ public void executeFuncAndSql(List<SqlBinds> mainStatementsBinds,
             logger.error("execute sql error: {}", e.getMessage());
             throw e;
         } finally {
-            close(resultSet, connection);
+            close(resultSet, statement, connection);
         }
     }
 
@@ -292,11 +295,9 @@ private void sendAttachment(int groupId, String title, String content) {
         setTaskAlertInfo(taskAlertInfo);
     }
 
-    private ResultSet executeQuery(Connection connection, SqlBinds sqlBinds, String handlerType) throws Exception {
-        try (PreparedStatement statement = prepareStatementAndBind(connection, sqlBinds)) {
-            logger.info("{} statement execute query, for sql: {}", handlerType, sqlBinds.getSql());
-            return statement.executeQuery();
-        }
+    private ResultSet executeQuery(SqlBinds sqlBinds, String handlerType, PreparedStatement statement) throws Exception {
+        logger.info("{} statement execute query, for sql: {}", handlerType, sqlBinds.getSql());
+        return statement.executeQuery();
     }
 
     private String executeUpdate(Connection connection, List<SqlBinds> statementsBinds, String handlerType) throws Exception {
@@ -332,7 +333,7 @@ private void createTempFunction(Connection connection,
      * @param resultSet resultSet
      * @param connection connection
      */
-    private void close(ResultSet resultSet, Connection connection) {
+    private void close(ResultSet resultSet, PreparedStatement pstmt, Connection connection) {
         if (resultSet != null) {
             try {
                 resultSet.close();
@@ -341,6 +342,14 @@ private void close(ResultSet resultSet, Connection connection) {
             }
         }
 
+        if (pstmt != null) {
+            try {
+                pstmt.close();
+            } catch (SQLException e) {
+                logger.error("close prepared statement error : {}", e.getMessage(), e);
+            }
+        }
+
         if (connection != null) {
             try {
                 connection.close();
