Browse Source

Fix bug: Use try-with-resources or close this "Socket" in a "finally" clause. (#1716)

* #1714
pull/2/head
Jave-Chen 5 years ago committed by Tboy
parent
commit
d371745ddf
  1. 20
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/FourLetterWordMain.java
  2. 40
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/sql/SqlTask.java

20
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) public static String send4LetterWord(String host, int port, String cmd, int timeout)
throws IOException { throws IOException {
LOG.info("connecting to " + host + " " + port); LOG.info("connecting to {} {}", host, port);
Socket sock = new Socket();
InetSocketAddress hostaddress= host != null ? new InetSocketAddress(host, port) : InetSocketAddress hostaddress= host != null ? new InetSocketAddress(host, port) :
new InetSocketAddress(InetAddress.getByName(null), 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.setSoTimeout(timeout);
sock.connect(hostaddress, timeout); sock.connect(hostaddress, timeout);
OutputStream outstream = sock.getOutputStream();
outstream.write(cmd.getBytes()); outstream.write(cmd.getBytes());
outstream.flush(); outstream.flush();
// this replicates NC - close the output stream before reading // this replicates NC - close the output stream before reading
sock.shutdownOutput(); sock.shutdownOutput();
reader =
new BufferedReader(
new InputStreamReader(sock.getInputStream()));
StringBuilder sb = new StringBuilder(); StringBuilder sb = new StringBuilder();
String line; String line;
while((line = reader.readLine()) != null) { while((line = reader.readLine()) != null) {
@ -84,11 +83,6 @@ public class FourLetterWordMain {
return sb.toString(); return sb.toString();
} catch (SocketTimeoutException e) { } catch (SocketTimeoutException e) {
throw new IOException("Exception while executing four letter word: " + cmd, e); throw new IOException("Exception while executing four letter word: " + cmd, e);
} finally {
sock.close();
if (reader != null) {
reader.close();
}
} }
} }
} }

40
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 // set the name of the current thread
String threadLoggerInfoName = String.format(Constants.TASK_LOG_INFO_FORMAT, taskProps.getTaskAppId()); String threadLoggerInfoName = String.format(Constants.TASK_LOG_INFO_FORMAT, taskProps.getTaskAppId());
Thread.currentThread().setName(threadLoggerInfoName); Thread.currentThread().setName(threadLoggerInfoName);
logger.info(sqlParameters.toString()); logger.info("Full sql parameters: {}", sqlParameters);
logger.info("sql type : {}, datasource : {}, sql : {} , localParams : {},udfs : {},showType : {},connParams : {}", logger.info("sql type : {}, datasource : {}, sql : {} , localParams : {},udfs : {},showType : {},connParams : {}",
sqlParameters.getType(), sqlParameters.getType(),
sqlParameters.getDatasource(), 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 // decide whether to executeQuery or executeUpdate based on sqlType
if (sqlParameters.getSqlType() == SqlType.QUERY.ordinal()) { if (sqlParameters.getSqlType() == SqlType.QUERY.ordinal()) {
// query statements need to be convert to JsonArray and inserted into Alert to send // query statements need to be convert to JsonArray and inserted into Alert to send
JSONArray resultJSONArray = new JSONArray(); JSONArray resultJSONArray = new JSONArray();
ResultSet resultSet = stmt.executeQuery();
ResultSetMetaData md = resultSet.getMetaData(); ResultSetMetaData md = resultSet.getMetaData();
int num = md.getColumnCount(); int num = md.getColumnCount();
@ -305,11 +305,10 @@ public class SqlTask extends AbstractTask {
} }
resultJSONArray.add(mapOfColValues); resultJSONArray.add(mapOfColValues);
} }
resultSet.close();
logger.debug("execute sql : {}", JSONObject.toJSONString(resultJSONArray, SerializerFeature.WriteMapNullValue)); logger.debug("execute sql : {}", JSONObject.toJSONString(resultJSONArray, SerializerFeature.WriteMapNullValue));
// if there is a result set // if there is a result set
if (resultJSONArray.size() > 0) { if ( !resultJSONArray.isEmpty() ) {
if (StringUtils.isNotEmpty(sqlParameters.getTitle())) { if (StringUtils.isNotEmpty(sqlParameters.getTitle())) {
sendAttachment(sqlParameters.getTitle(), sendAttachment(sqlParameters.getTitle(),
JSONObject.toJSONString(resultJSONArray, SerializerFeature.WriteMapNullValue)); JSONObject.toJSONString(resultJSONArray, SerializerFeature.WriteMapNullValue));
@ -337,6 +336,12 @@ public class SqlTask extends AbstractTask {
} catch (Exception e) { } catch (Exception e) {
logger.error(e.getMessage(),e); logger.error(e.getMessage(),e);
throw new RuntimeException(e.getMessage()); throw new RuntimeException(e.getMessage());
} finally {
try {
connection.close();
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
} }
return connection; return connection;
} }
@ -349,22 +354,23 @@ public class SqlTask extends AbstractTask {
* @throws Exception * @throws Exception
*/ */
private PreparedStatement prepareStatementAndBind(Connection connection, SqlBinds sqlBinds) throws Exception { private PreparedStatement prepareStatementAndBind(Connection connection, SqlBinds sqlBinds) throws Exception {
PreparedStatement stmt = connection.prepareStatement(sqlBinds.getSql());
// is the timeout set // is the timeout set
boolean timeoutFlag = taskProps.getTaskTimeoutStrategy() == TaskTimeoutStrategy.FAILED || boolean timeoutFlag = taskProps.getTaskTimeoutStrategy() == TaskTimeoutStrategy.FAILED ||
taskProps.getTaskTimeoutStrategy() == TaskTimeoutStrategy.WARNFAILED; taskProps.getTaskTimeoutStrategy() == TaskTimeoutStrategy.WARNFAILED;
if(timeoutFlag){ try (PreparedStatement stmt = connection.prepareStatement(sqlBinds.getSql())) {
stmt.setQueryTimeout(taskProps.getTaskTimeout()); if(timeoutFlag){
} stmt.setQueryTimeout(taskProps.getTaskTimeout());
Map<Integer, Property> params = sqlBinds.getParamsMap(); }
if(params != null) { Map<Integer, Property> params = sqlBinds.getParamsMap();
for (Map.Entry<Integer, Property> entry : params.entrySet()) { if(params != null) {
Property prop = entry.getValue(); for (Map.Entry<Integer, Property> entry : params.entrySet()) {
ParameterUtils.setInParameter(entry.getKey(), stmt, prop.getType(), prop.getValue()); 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++){ for(int i=1;i<=sqlParamsMap.size();i++){
logPrint.append(sqlParamsMap.get(i).getValue()+"("+sqlParamsMap.get(i).getType()+")"); logPrint.append(sqlParamsMap.get(i).getValue()+"("+sqlParamsMap.get(i).getType()+")");
} }
logger.info(logPrint.toString()); logger.info("Sql Params are {}", logPrint);
} }
/** /**

Loading…
Cancel
Save