Merge pull request #5 from apache/dev-db

Dev db
apache · Oct 21, 2019 · 14f6ab4 · 14f6ab4
2 parents 9f587a5 + 26c96d8
commit 14f6ab4
Show file tree

Hide file tree

Showing 18 changed files with 63 additions and 28 deletions.
diff --git a/...er-alert/src/main/java/org/apache/dolphinscheduler/alert/utils/EnterpriseWeChatUtils.java b/...er-alert/src/main/java/org/apache/dolphinscheduler/alert/utils/EnterpriseWeChatUtils.java
@@ -214,7 +214,7 @@ public static String markdownText(String title,String content){
             }
 
             StringBuilder contents = new StringBuilder(100);
-            contents.append(String.format("`%s`\n",title));
+            contents.append(String.format("`%s`%n",title));
             for (String str : list){
                 contents.append(Constants.MARKDOWN_QUOTE);
                 contents.append(str);

diff --git a/...nscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/utils/PropertyUtils.java b/...nscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/utils/PropertyUtils.java
@@ -56,6 +56,9 @@ private void init(){
 
             } catch (IOException e) {
                 logger.error(e.getMessage(), e);
+                if (fis != null) {
+                    IOUtils.closeQuietly(fis);
+                }
                 System.exit(1);
             } finally {
                 IOUtils.closeQuietly(fis);
@@ -121,7 +124,7 @@ public static Boolean getBoolean(String key) {
             return Boolean.parseBoolean(value);
         }
 
-        return null;
+        return false;
     }
 
     /**

diff --git a/...r-api/src/main/java/org/apache/dolphinscheduler/api/controller/WorkerGroupController.java b/...r-api/src/main/java/org/apache/dolphinscheduler/api/controller/WorkerGroupController.java
@@ -110,7 +110,7 @@ public Result queryAllWorkerGroupsPaging(@ApiIgnore @RequestAttribute(value = Co
 
         try {
             searchVal = ParameterUtils.handleEscapes(searchVal);
-            Map<String, Object> result = workerGroupService.queryAllGroupPaging(pageNo, pageSize, searchVal);
+            Map<String, Object> result = workerGroupService.queryAllGroupPaging(loginUser,pageNo, pageSize, searchVal);
             return returnDataListPaging(result);
         }catch (Exception e){
             logger.error(Status.SAVE_ERROR.getMsg(),e);

diff --git a/...heduler-api/src/main/java/org/apache/dolphinscheduler/api/service/AccessTokenService.java b/...heduler-api/src/main/java/org/apache/dolphinscheduler/api/service/AccessTokenService.java
@@ -57,6 +57,9 @@ public class AccessTokenService extends BaseService {
      */
     public Map<String, Object> queryAccessTokenList(User loginUser, String searchVal, Integer pageNo, Integer pageSize) {
         Map<String, Object> result = new HashMap<>(5);
+        if (checkAdmin(loginUser, result)) {
+            return result;
+        }
 
         PageInfo<AccessToken> pageInfo = new PageInfo<>(pageNo, pageSize);
         Page<AccessToken> page = new Page(pageNo, pageSize);

diff --git a/...cheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/AlertGroupService.java b/...cheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/AlertGroupService.java
@@ -81,6 +81,9 @@ public HashMap<String, Object> queryAlertgroup() {
     public Map<String, Object> listPaging(User loginUser, String searchVal, Integer pageNo, Integer pageSize) {
 
         Map<String, Object> result = new HashMap<>(5);
+        if (checkAdmin(loginUser, result)) {
+            return result;
+        }
 
         Page<AlertGroup> page = new Page(pageNo, pageSize);
         IPage<AlertGroup> alertGroupIPage = alertGroupMapper.queryAlertGroupPage(

diff --git a/...r-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionService.java b/...r-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionService.java
@@ -870,7 +870,7 @@ public Map<String, Object> getTaskNodeListByDefinitionId(Integer defineId) throw
         ProcessDefinition processDefinition = processDefineMapper.selectById(defineId);
         if (processDefinition == null) {
             logger.info("process define not exists");
-            putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, processDefinition.getId());
+            putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, defineId);
             return result;
         }
 

diff --git a/...ler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessInstanceService.java b/...ler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessInstanceService.java
@@ -120,9 +120,9 @@ public Map<String, Object> queryProcessInstanceById(User loginUser, String proje
         }else{
             WorkerGroup workerGroup = workerGroupMapper.selectById(processInstance.getWorkerGroupId());
             if(workerGroup != null){
-                workerGroupName = DEFAULT;
-            }else{
                 workerGroupName = workerGroup.getName();
+            }else{
+                workerGroupName = DEFAULT;
             }
         }
         processInstance.setWorkerGroupName(workerGroupName);
@@ -165,14 +165,10 @@ public Map<String, Object> queryProcessInstanceList(User loginUser, String proje
         }
 
         int[] statusArray = null;
-        String statesStr = null;
         // filter by state
         if (stateType != null) {
             statusArray = new int[]{stateType.ordinal()};
         }
-        if (statusArray != null) {
-            statesStr = Arrays.toString(statusArray).replace("[", "").replace("]", "");
-        }
 
         Date start = null;
         Date end = null;

diff --git a/...heduler-api/src/main/java/org/apache/dolphinscheduler/api/service/WorkerGroupService.java b/...heduler-api/src/main/java/org/apache/dolphinscheduler/api/service/WorkerGroupService.java
@@ -19,6 +19,7 @@
 import org.apache.dolphinscheduler.api.enums.Status;
 import org.apache.dolphinscheduler.api.utils.Constants;
 import org.apache.dolphinscheduler.api.utils.PageInfo;
+import org.apache.dolphinscheduler.dao.entity.User;
 import org.apache.dolphinscheduler.dao.entity.WorkerGroup;
 import org.apache.dolphinscheduler.dao.mapper.WorkerGroupMapper;
 import com.baomidou.mybatisplus.core.metadata.IPage;
@@ -108,14 +109,18 @@ private boolean checkWorkerGroupNameExists(WorkerGroup workerGroup) {
 
     /**
      * query worker group paging
+     * @param loginUser
      * @param pageNo
      * @param pageSize
      * @param searchVal
      * @return
      */
-    public Map<String,Object> queryAllGroupPaging(Integer pageNo, Integer pageSize, String searchVal) {
+    public Map<String,Object> queryAllGroupPaging(User loginUser, Integer pageNo, Integer pageSize, String searchVal) {
 
         Map<String, Object> result = new HashMap<>(5);
+        if (checkAdmin(loginUser, result)) {
+            return result;
+        }
 
         Page<WorkerGroup> page = new Page(pageNo, pageSize);
         IPage<WorkerGroup> workerGroupIPage = workerGroupMapper.queryListPaging(

diff --git a/...duler-common/src/main/java/org/apache/dolphinscheduler/common/model/TaskNodeRelation.java b/...duler-common/src/main/java/org/apache/dolphinscheduler/common/model/TaskNodeRelation.java
@@ -53,8 +53,12 @@ public void setEndNode(String endNode) {
   }
 
 
-  public boolean equals(TaskNodeRelation e){
-      return (e.getStartNode() == this.startNode && e.getEndNode() == this.endNode);
+  public boolean equals(Object o){
+    if (!(o instanceof TaskNodeRelation)) {
+      return false;
+    }
+    TaskNodeRelation relation = (TaskNodeRelation)o;
+    return (relation.getStartNode().equals(this.startNode) && relation.getEndNode().equals(this.endNode));
   }
 
   @Override

diff --git a/...hinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/FileUtils.java b/...hinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/FileUtils.java
@@ -153,8 +153,9 @@ public static boolean writeContent2File(String content, String filePath) {
         BufferedWriter bufferedWriter = null;
         try {
             File distFile = new File(filePath);
-            if (!distFile.getParentFile().exists()) {
-                distFile.getParentFile().mkdirs();
+            if (!distFile.getParentFile().exists() && !distFile.getParentFile().mkdirs()) {
+                FileUtils.logger.error("mkdir parent failed");
+                return false;
             }
             bufferedReader = new BufferedReader(new StringReader(content));
             bufferedWriter = new BufferedWriter(new FileWriter(distFile));

diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/OSUtils.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/OSUtils.java
@@ -156,7 +156,9 @@ public static List<String> getUserList() {
       logger.error(e.getMessage(), e);
     } finally {
       try {
-        bufferedReader.close();
+        if (bufferedReader != null) {
+          bufferedReader.close();
+        }
       } catch (IOException e) {
         logger.error(e.getMessage(), e);
       }

diff --git a/...cheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/PropertyUtils.java b/...cheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/PropertyUtils.java
@@ -60,6 +60,9 @@ private void init(){
 
             } catch (IOException e) {
                 logger.error(e.getMessage(), e);
+                if (fis != null) {
+                    IOUtils.closeQuietly(fis);
+                }
                 System.exit(1);
             } finally {
                 IOUtils.closeQuietly(fis);
@@ -173,15 +176,15 @@ public double getDouble(String key, double defaultVal) {
     public static String[] getArray(String key, String splitStr) {
         String value = getString(key);
         if (value == null) {
-            return null;
+            return new String[0];
         }
         try {
             String[] propertyArray = value.split(splitStr);
             return propertyArray;
         } catch (NumberFormatException e) {
             logger.info(e.getMessage(),e);
         }
-        return null;
+        return new String[0];
     }
 
     /**

diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/MonitorDBDao.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/MonitorDBDao.java
@@ -83,6 +83,7 @@ public static List<MonitorRecord> queryDatabaseState() {
         List<MonitorRecord> list = new ArrayList<>(1);
 
         Connection conn = null;
+        Statement pstmt = null;
         long maxConnections = 0;
         long maxUsedConnections = 0;
         long threadsConnections = 0;
@@ -98,7 +99,7 @@ public static List<MonitorRecord> queryDatabaseState() {
                 return list;
             }
 
-            Statement pstmt = conn.createStatement();
+            pstmt = conn.createStatement();
 
             ResultSet rs1 = pstmt.executeQuery("show global variables");
             while(rs1.next()){
@@ -124,6 +125,9 @@ public static List<MonitorRecord> queryDatabaseState() {
             state = 0;
         }finally {
             try {
+                if(pstmt != null) {
+                    pstmt.close();
+                }
                 if(conn != null){
                     conn.close();
                 }

diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/TaskRecordDao.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/TaskRecordDao.java
@@ -147,14 +147,14 @@ public static int countTaskRecord(Map<String, String> filterMap, String table){
 
         int count = 0;
         Connection conn = null;
+        PreparedStatement pstmt = null;
         try {
             conn = getConn();
             if(conn == null){
                 return count;
             }
             String sql = String.format("select count(1) as count from %s", table);
             sql += getWhereString(filterMap);
-            PreparedStatement pstmt;
             pstmt = conn.prepareStatement(sql);
             ResultSet rs = pstmt.executeQuery();
             while(rs.next()){
@@ -165,6 +165,9 @@ public static int countTaskRecord(Map<String, String> filterMap, String table){
             logger.error("Exception ", e);
         }finally {
             try {
+                if(pstmt != null) {
+                    pstmt.close();
+                }
                 if(conn != null){
                     conn.close();
                 }
@@ -234,12 +237,12 @@ private static TaskRecord convertToTaskRecord(ResultSet resultSet) throws SQLExc
     private static List<TaskRecord> getQueryResult(String selectSql) {
         List<TaskRecord> recordList = new ArrayList<>();
         Connection conn = null;
+        PreparedStatement pstmt = null;
         try {
             conn = getConn();
             if(conn == null){
                 return recordList;
             }
-            PreparedStatement pstmt;
             pstmt = conn.prepareStatement(selectSql);
             ResultSet rs = pstmt.executeQuery();
 
@@ -251,6 +254,9 @@ private static List<TaskRecord> getQueryResult(String selectSql) {
             logger.error("Exception ", e);
         }finally {
             try {
+                if(pstmt != null) {
+                    pstmt.close();
+                }
                 if(conn != null){
                     conn.close();
                 }

diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/PropertyUtils.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/PropertyUtils.java
@@ -55,6 +55,9 @@ private void init(){
 
             } catch (IOException e) {
                 logger.error(e.getMessage(), e);
+                if (fis != null) {
+                    IOUtils.closeQuietly(fis);
+                }
                 System.exit(1);
             } finally {
                 IOUtils.closeQuietly(fis);
@@ -120,7 +123,7 @@ public static Boolean getBoolean(String key) {
             return Boolean.parseBoolean(value);
         }
 
-        return null;
+        return false;
     }
 
     /**

diff --git a/.../src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterTaskExecThread.java b/.../src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterTaskExecThread.java
@@ -74,7 +74,6 @@ public Boolean submitWaitComplete() {
     public Boolean waitTaskQuit(){
         // query new state
         taskInstance = processDao.findTaskInstanceById(taskInstance.getId());
-        Boolean result = true;
         // task time out
         Boolean checkTimeout = false;
         TaskTimeoutParameter taskTimeoutParameter = getTaskTimeoutParameter();
@@ -89,7 +88,7 @@ public Boolean waitTaskQuit(){
             try {
                 if(this.processInstance == null){
                     logger.error("process instance not exists , master task exec thread exit");
-                    return result;
+                    return true;
                 }
                 // task instance add queue , waiting worker to kill
                 if(this.cancel || this.processInstance.getState() == ExecutionStatus.READY_STOP){
@@ -116,11 +115,13 @@ public Boolean waitTaskQuit(){
                 Thread.sleep(Constants.SLEEP_TIME_MILLIS);
             } catch (Exception e) {
                 logger.error("exception: "+ e.getMessage(),e);
-                logger.error("wait task quit failed, instance id:{}, task id:{}",
-                        processInstance.getId(), taskInstance.getId());
+                if (processInstance != null) {
+                    logger.error("wait task quit failed, instance id:{}, task id:{}",
+                            processInstance.getId(), taskInstance.getId());
+                }
             }
         }
-        return  result;
+        return true;
     }
 
 

diff --git a/...ler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/AbstractTask.java b/...ler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/AbstractTask.java
@@ -181,6 +181,7 @@ private Class getCurTaskParamsClass(){
                 break;
             case FLINK:
                 paramsClass = FlinkParameters.class;
+                break;
             case PYTHON:
                 paramsClass = PythonParameters.class;
                 break;

diff --git a/...uler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/sql/SqlTask.java b/...uler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/sql/SqlTask.java
@@ -303,7 +303,7 @@ public Connection executeFuncAndSql(SqlBinds mainSqlBinds,
                         }
                         resultJSONArray.add(mapOfColValues);
                     }
-
+                    resultSet.close();
                     logger.debug("execute sql : {}", JSONObject.toJSONString(resultJSONArray, SerializerFeature.WriteMapNullValue));
 
                     // if there is a result set