diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/FourLetterWordMain.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/FourLetterWordMain.java index d41830eff5..b04e773aea 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/FourLetterWordMain.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/FourLetterWordMain.java @@ -59,23 +59,22 @@ public class FourLetterWordMain { */ public static String send4LetterWord(String host, int port, String cmd, int timeout) throws IOException { - LOG.info("connecting to " + host + " " + port); - Socket sock = new Socket(); + LOG.info("connecting to {} {}", host, port); InetSocketAddress hostaddress= host != null ? new InetSocketAddress(host, port) : new InetSocketAddress(InetAddress.getByName(null), port); - BufferedReader reader = null; - try { + + try (Socket sock = new Socket(); + OutputStream outstream = sock.getOutputStream(); + BufferedReader reader = + new BufferedReader( + new InputStreamReader(sock.getInputStream()))) { sock.setSoTimeout(timeout); sock.connect(hostaddress, timeout); - OutputStream outstream = sock.getOutputStream(); outstream.write(cmd.getBytes()); outstream.flush(); // this replicates NC - close the output stream before reading sock.shutdownOutput(); - reader = - new BufferedReader( - new InputStreamReader(sock.getInputStream())); StringBuilder sb = new StringBuilder(); String line; while((line = reader.readLine()) != null) { @@ -84,11 +83,6 @@ public class FourLetterWordMain { return sb.toString(); } catch (SocketTimeoutException e) { throw new IOException("Exception while executing four letter word: " + cmd, e); - } finally { - sock.close(); - if (reader != null) { - reader.close(); - } } } } diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/sql/SqlTask.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/sql/SqlTask.java index e8a97fecc5..a2f07c8be3 100644 --- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/sql/SqlTask.java +++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/sql/SqlTask.java @@ -105,7 +105,7 @@ public class SqlTask extends AbstractTask { // set the name of the current thread String threadLoggerInfoName = String.format(Constants.TASK_LOG_INFO_FORMAT, taskProps.getTaskAppId()); Thread.currentThread().setName(threadLoggerInfoName); - logger.info(sqlParameters.toString()); + logger.info("Full sql parameters: {}", sqlParameters); logger.info("sql type : {}, datasource : {}, sql : {} , localParams : {},udfs : {},showType : {},connParams : {}", sqlParameters.getType(), sqlParameters.getDatasource(), @@ -289,12 +289,12 @@ public class SqlTask extends AbstractTask { } } - try (PreparedStatement stmt = prepareStatementAndBind(connection, mainSqlBinds)) { + try (PreparedStatement stmt = prepareStatementAndBind(connection, mainSqlBinds); + ResultSet resultSet = stmt.executeQuery()) { // 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 JSONArray resultJSONArray = new JSONArray(); - ResultSet resultSet = stmt.executeQuery(); ResultSetMetaData md = resultSet.getMetaData(); int num = md.getColumnCount(); @@ -305,11 +305,10 @@ public class SqlTask extends AbstractTask { } resultJSONArray.add(mapOfColValues); } - resultSet.close(); logger.debug("execute sql : {}", JSONObject.toJSONString(resultJSONArray, SerializerFeature.WriteMapNullValue)); // if there is a result set - if (resultJSONArray.size() > 0) { + if ( !resultJSONArray.isEmpty() ) { if (StringUtils.isNotEmpty(sqlParameters.getTitle())) { sendAttachment(sqlParameters.getTitle(), JSONObject.toJSONString(resultJSONArray, SerializerFeature.WriteMapNullValue)); @@ -337,6 +336,12 @@ public class SqlTask extends AbstractTask { } catch (Exception e) { logger.error(e.getMessage(),e); throw new RuntimeException(e.getMessage()); + } finally { + try { + connection.close(); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } } return connection; } @@ -349,22 +354,23 @@ public class SqlTask extends AbstractTask { * @throws Exception */ private PreparedStatement prepareStatementAndBind(Connection connection, SqlBinds sqlBinds) throws Exception { - PreparedStatement stmt = connection.prepareStatement(sqlBinds.getSql()); // is the timeout set boolean timeoutFlag = taskProps.getTaskTimeoutStrategy() == TaskTimeoutStrategy.FAILED || taskProps.getTaskTimeoutStrategy() == TaskTimeoutStrategy.WARNFAILED; - if(timeoutFlag){ - stmt.setQueryTimeout(taskProps.getTaskTimeout()); - } - Map params = sqlBinds.getParamsMap(); - if(params != null) { - for (Map.Entry entry : params.entrySet()) { - Property prop = entry.getValue(); - ParameterUtils.setInParameter(entry.getKey(), stmt, prop.getType(), prop.getValue()); + try (PreparedStatement stmt = connection.prepareStatement(sqlBinds.getSql())) { + if(timeoutFlag){ + stmt.setQueryTimeout(taskProps.getTaskTimeout()); + } + Map params = sqlBinds.getParamsMap(); + if(params != null) { + for (Map.Entry entry : params.entrySet()) { + Property prop = entry.getValue(); + ParameterUtils.setInParameter(entry.getKey(), stmt, prop.getType(), prop.getValue()); + } } + logger.info("prepare statement replace sql : {} ", stmt); + return stmt; } - logger.info("prepare statement replace sql : {} ",stmt.toString()); - return stmt; } /** @@ -452,7 +458,7 @@ public class SqlTask extends AbstractTask { for(int i=1;i<=sqlParamsMap.size();i++){ logPrint.append(sqlParamsMap.get(i).getValue()+"("+sqlParamsMap.get(i).getType()+")"); } - logger.info(logPrint.toString()); + logger.info("Sql Params are {}", logPrint); } /**