You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by ga...@apache.org on 2020/12/31 02:19:05 UTC
[incubator-dolphinscheduler] branch alert_plugin_design updated: [Feature-#3828][server] Delete recipients and CCs in the process defi… (#4303)
This is an automated email from the ASF dual-hosted git repository.
gaojun2048 pushed a commit to branch alert_plugin_design
in repository https://gitbox.apache.org/repos/asf/incubator-dolphinscheduler.git
The following commit(s) were added to refs/heads/alert_plugin_design by this push:
new 14785f4 [Feature-#3828][server] Delete recipients and CCs in the process defi… (#4303)
14785f4 is described below
commit 14785f4e390e3f2dcbfa223bd33dbc1907f280d3
Author: Kirs <ac...@163.com>
AuthorDate: Thu Dec 31 10:18:53 2020 +0800
[Feature-#3828][server] Delete recipients and CCs in the process defi… (#4303)
* [Feature-#3828][server] Delete recipients and CCs in the process definition and select the associated alert group instead
* The specific alarm instance associated with the alert group
* The process definition only associates the alert group, and does not care about the specific alarm instance.
this closes #3828
* remove test cc
* remove test cc
* fix test error
* add sql upgrade script
delete alert column(show_type、alert_type、re、cc)
* code style
* update test
Note the statement to delete the column, and provide related instructions
* delete AlertGroup field(alert_type)
* fix ut error
* fix ut error
* fix ut error
* code style
* delete api-doc receiverCC
* delete UserAlertGroupMapper
* delete UserAlertGroup
* delete UserAlertGroupApi
* alert group add coulmn user_id
* alter user_id to create_user_id
* userinfo remove alert group info
---
.../alert/plugin/EmailAlertPluginTest.java | 3 -
.../api/controller/AlertGroupController.java | 91 ++-
.../api/controller/ExecutorController.java | 116 ++--
.../api/service/AlertGroupService.java | 94 +--
.../api/service/ExecutorService.java | 47 +-
.../api/service/ProcessInstanceService.java | 3 +-
.../api/service/SchedulerService.java | 7 +-
.../api/service/UserAlertGroupService.java | 38 -
.../dolphinscheduler/api/service/UsersService.java | 178 +++--
.../service/impl/ProcessDefinitionServiceImpl.java | 32 +-
.../impl/ProcessDefinitionVersionServiceImpl.java | 23 +-
.../api/controller/ExecutorControllerTest.java | 100 ++-
.../api/service/AlertGroupServiceTest.java | 129 ++--
.../api/service/ExecutorService2Test.java | 126 ++--
.../ProcessDefinitionVersionServiceTest.java | 23 +-
.../api/service/UserAlertGroupServiceTest.java | 53 --
.../org/apache/dolphinscheduler/dao/AlertDao.java | 85 +--
.../apache/dolphinscheduler/dao/entity/Alert.java | 85 +--
.../dolphinscheduler/dao/entity/AlertGroup.java | 58 +-
.../dao/entity/ProcessDefinition.java | 93 ++-
.../dao/entity/ProcessDefinitionVersion.java | 88 +--
.../dao/entity/UserAlertGroup.java | 142 ----
.../dao/mapper/AlertGroupMapper.java | 8 -
.../dao/mapper/UserAlertGroupMapper.java | 45 --
.../dao/mapper/AlertGroupMapper.xml | 8 +-
.../dao/mapper/ProcessDefinitionMapper.xml | 12 +-
.../dao/mapper/UserAlertGroupMapper.xml | 31 -
.../apache/dolphinscheduler/dao/AlertDaoTest.java | 7 +-
.../dao/mapper/AlertGroupMapperTest.java | 189 ++---
.../dao/mapper/AlertMapperTest.java | 5 -
.../dao/mapper/AlertPluginInstanceMapperTest.java | 2 -
.../dao/mapper/ProcessDefinitionMapperTest.java | 4 -
.../mapper/ProcessDefinitionVersionMapperTest.java | 4 -
.../dao/mapper/UserAlertGroupMapperTest.java | 196 ------
.../dao/mapper/UserMapperTest.java | 122 +---
.../server/master/runner/MasterTaskExecThread.java | 77 ++-
.../server/utils/AlertManager.java | 12 -
.../server/registry/DependencyConfig.java | 7 -
.../processor/TaskCallbackServiceTestConfig.java | 8 -
.../service/process/ProcessService.java | 766 +++++++++++----------
pom.xml | 1 -
sql/dolphinscheduler-postgre.sql | 63 +-
sql/dolphinscheduler_mysql.sql | 66 +-
.../1.4.0_schema/mysql/dolphinscheduler_ddl.sql | 42 +-
.../postgresql/dolphinscheduler_ddl.sql | 40 +-
45 files changed, 1189 insertions(+), 2140 deletions(-)
diff --git a/dolphinscheduler-alert/src/test/java/org/apache/dolphinscheduler/alert/plugin/EmailAlertPluginTest.java b/dolphinscheduler-alert/src/test/java/org/apache/dolphinscheduler/alert/plugin/EmailAlertPluginTest.java
index a6bd51c..183a805 100644
--- a/dolphinscheduler-alert/src/test/java/org/apache/dolphinscheduler/alert/plugin/EmailAlertPluginTest.java
+++ b/dolphinscheduler-alert/src/test/java/org/apache/dolphinscheduler/alert/plugin/EmailAlertPluginTest.java
@@ -22,7 +22,6 @@ import org.apache.dolphinscheduler.alert.runner.AlertSender;
import org.apache.dolphinscheduler.alert.utils.Constants;
import org.apache.dolphinscheduler.alert.utils.PropertyUtils;
import org.apache.dolphinscheduler.common.enums.AlertStatus;
-import org.apache.dolphinscheduler.common.enums.AlertType;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.AlertDao;
import org.apache.dolphinscheduler.dao.DaoFactory;
@@ -70,8 +69,6 @@ public class EmailAlertPluginTest {
AlertGroup alertGroup = new AlertGroup();
alertGroup.setDescription("test alert group 1");
alertGroup.setGroupName("testalertg1");
- //TODO AlertType is neet delete from AlertGroup
- alertGroup.setGroupType(AlertType.EMAIL);
alertDao.getAlertGroupMapper().insert(alertGroup);
//add alert
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/AlertGroupController.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/AlertGroupController.java
index 35bbc2a..b3a3ee5 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/AlertGroupController.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/AlertGroupController.java
@@ -14,8 +14,16 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.api.controller;
+import static org.apache.dolphinscheduler.api.enums.Status.CREATE_ALERT_GROUP_ERROR;
+import static org.apache.dolphinscheduler.api.enums.Status.DELETE_ALERT_GROUP_ERROR;
+import static org.apache.dolphinscheduler.api.enums.Status.LIST_PAGING_ALERT_GROUP_ERROR;
+import static org.apache.dolphinscheduler.api.enums.Status.QUERY_ALL_ALERTGROUP_ERROR;
+import static org.apache.dolphinscheduler.api.enums.Status.UPDATE_ALERT_GROUP_ERROR;
+
+import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.exceptions.ApiException;
import org.apache.dolphinscheduler.api.service.AlertGroupService;
import org.apache.dolphinscheduler.api.utils.Result;
@@ -23,22 +31,27 @@ import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.AlertType;
import org.apache.dolphinscheduler.common.utils.ParameterUtils;
import org.apache.dolphinscheduler.dao.entity.User;
-import io.swagger.annotations.Api;
-import io.swagger.annotations.ApiImplicitParam;
-import io.swagger.annotations.ApiImplicitParams;
-import io.swagger.annotations.ApiOperation;
-import org.apache.dolphinscheduler.api.enums.Status;
+
+import java.util.HashMap;
+import java.util.Map;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
-import org.springframework.web.bind.annotation.*;
-import springfox.documentation.annotations.ApiIgnore;
-
-import java.util.HashMap;
-import java.util.Map;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestAttribute;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseStatus;
+import org.springframework.web.bind.annotation.RestController;
-import static org.apache.dolphinscheduler.api.enums.Status.*;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiImplicitParam;
+import io.swagger.annotations.ApiImplicitParams;
+import io.swagger.annotations.ApiOperation;
+import springfox.documentation.annotations.ApiIgnore;
/**
* alert group controller
@@ -65,9 +78,9 @@ public class AlertGroupController extends BaseController {
*/
@ApiOperation(value = "createAlertgroup", notes = "CREATE_ALERT_GROUP_NOTES")
@ApiImplicitParams({
- @ApiImplicitParam(name = "groupName", value = "GROUP_NAME", required = true, dataType = "String"),
- @ApiImplicitParam(name = "groupType", value = "GROUP_TYPE", required = true, dataType = "AlertType"),
- @ApiImplicitParam(name = "description", value = "DESC", dataType = "String")
+ @ApiImplicitParam(name = "groupName", value = "GROUP_NAME", required = true, dataType = "String"),
+ @ApiImplicitParam(name = "groupType", value = "GROUP_TYPE", required = true, dataType = "AlertType"),
+ @ApiImplicitParam(name = "description", value = "DESC", dataType = "String")
})
@PostMapping(value = "/create")
@ResponseStatus(HttpStatus.CREATED)
@@ -77,7 +90,7 @@ public class AlertGroupController extends BaseController {
@RequestParam(value = "groupType") AlertType groupType,
@RequestParam(value = "description", required = false) String description) {
logger.info("loginUser user {}, create alertgroup, groupName: {}, groupType: {}, desc: {}",
- loginUser.getUserName(), groupName, groupType, description);
+ loginUser.getUserName(), groupName, groupType, description);
Map<String, Object> result = alertGroupService.createAlertgroup(loginUser, groupName, groupType, description);
return returnDataList(result);
}
@@ -94,7 +107,7 @@ public class AlertGroupController extends BaseController {
@ApiException(QUERY_ALL_ALERTGROUP_ERROR)
public Result list(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser) {
logger.info("login user {}, query all alertGroup",
- loginUser.getUserName());
+ loginUser.getUserName());
HashMap<String, Object> result = alertGroupService.queryAlertgroup();
return returnDataList(result);
}
@@ -110,9 +123,9 @@ public class AlertGroupController extends BaseController {
*/
@ApiOperation(value = "queryAlertGroupListPaging", notes = "QUERY_ALERT_GROUP_LIST_PAGING_NOTES")
@ApiImplicitParams({
- @ApiImplicitParam(name = "searchVal", value = "SEARCH_VAL", type = "String"),
- @ApiImplicitParam(name = "pageNo", value = "PAGE_NO", dataType = "Int", example = "1"),
- @ApiImplicitParam(name = "pageSize", value = "PAGE_SIZE", dataType = "Int", example = "20")
+ @ApiImplicitParam(name = "searchVal", value = "SEARCH_VAL", type = "String"),
+ @ApiImplicitParam(name = "pageNo", value = "PAGE_NO", dataType = "Int", example = "1"),
+ @ApiImplicitParam(name = "pageSize", value = "PAGE_SIZE", dataType = "Int", example = "20")
})
@GetMapping(value = "/list-paging")
@ResponseStatus(HttpStatus.OK)
@@ -122,7 +135,7 @@ public class AlertGroupController extends BaseController {
@RequestParam(value = "searchVal", required = false) String searchVal,
@RequestParam("pageSize") Integer pageSize) {
logger.info("login user {}, list paging, pageNo: {}, searchVal: {}, pageSize: {}",
- loginUser.getUserName(), pageNo, searchVal, pageSize);
+ loginUser.getUserName(), pageNo, searchVal, pageSize);
Map<String, Object> result = checkPageParams(pageNo, pageSize);
if (result.get(Constants.STATUS) != Status.SUCCESS) {
return returnDataListPaging(result);
@@ -145,10 +158,10 @@ public class AlertGroupController extends BaseController {
*/
@ApiOperation(value = "updateAlertgroup", notes = "UPDATE_ALERT_GROUP_NOTES")
@ApiImplicitParams({
- @ApiImplicitParam(name = "id", value = "ALERT_GROUP_ID", required = true, dataType = "Int", example = "100"),
- @ApiImplicitParam(name = "groupName", value = "GROUP_NAME", required = true, dataType = "String"),
- @ApiImplicitParam(name = "groupType", value = "GROUP_TYPE", required = true, dataType = "AlertType"),
- @ApiImplicitParam(name = "description", value = "DESC", dataType = "String")
+ @ApiImplicitParam(name = "id", value = "ALERT_GROUP_ID", required = true, dataType = "Int", example = "100"),
+ @ApiImplicitParam(name = "groupName", value = "GROUP_NAME", required = true, dataType = "String"),
+ @ApiImplicitParam(name = "groupType", value = "GROUP_TYPE", required = true, dataType = "AlertType"),
+ @ApiImplicitParam(name = "description", value = "DESC", dataType = "String")
})
@PostMapping(value = "/update")
@ResponseStatus(HttpStatus.OK)
@@ -159,7 +172,7 @@ public class AlertGroupController extends BaseController {
@RequestParam(value = "groupType") AlertType groupType,
@RequestParam(value = "description", required = false) String description) {
logger.info("login user {}, updateProcessInstance alertgroup, groupName: {}, groupType: {}, desc: {}",
- loginUser.getUserName(), groupName, groupType, description);
+ loginUser.getUserName(), groupName, groupType, description);
Map<String, Object> result = alertGroupService.updateAlertgroup(loginUser, id, groupName, groupType, description);
return returnDataList(result);
}
@@ -173,7 +186,7 @@ public class AlertGroupController extends BaseController {
*/
@ApiOperation(value = "delAlertgroupById", notes = "DELETE_ALERT_GROUP_BY_ID_NOTES")
@ApiImplicitParams({
- @ApiImplicitParam(name = "id", value = "ALERT_GROUP_ID", required = true, dataType = "Int", example = "100")
+ @ApiImplicitParam(name = "id", value = "ALERT_GROUP_ID", required = true, dataType = "Int", example = "100")
})
@PostMapping(value = "/delete")
@ResponseStatus(HttpStatus.OK)
@@ -195,7 +208,7 @@ public class AlertGroupController extends BaseController {
*/
@ApiOperation(value = "verifyGroupName", notes = "VERIFY_ALERT_GROUP_NAME_NOTES")
@ApiImplicitParams({
- @ApiImplicitParam(name = "groupName", value = "GROUP_NAME", required = true, dataType = "String"),
+ @ApiImplicitParam(name = "groupName", value = "GROUP_NAME", required = true, dataType = "String"),
})
@GetMapping(value = "/verify-group-name")
@ResponseStatus(HttpStatus.OK)
@@ -215,28 +228,4 @@ public class AlertGroupController extends BaseController {
}
return result;
}
-
- /**
- * grant user
- *
- * @param loginUser login user
- * @param userIds user ids in the group
- * @param alertgroupId alert group id
- * @return grant result code
- */
- @ApiOperation(value = "grantUser", notes = "GRANT_ALERT_GROUP_NOTES")
- @ApiImplicitParams({
- @ApiImplicitParam(name = "id", value = "ALERT_GROUP_ID", required = true, dataType = "Int", example = "100"),
- @ApiImplicitParam(name = "userIds", value = "USER_IDS", required = true, dataType = "String")
- })
- @PostMapping(value = "/grant-user")
- @ResponseStatus(HttpStatus.OK)
- @ApiException(ALERT_GROUP_GRANT_USER_ERROR)
- public Result grantUser(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
- @RequestParam(value = "alertgroupId") int alertgroupId,
- @RequestParam(value = "userIds") String userIds) {
- logger.info("login user {}, grant user, alertGroupId: {},userIds : {}", loginUser.getUserName(), alertgroupId, userIds);
- Map<String, Object> result = alertGroupService.grantUser(loginUser, alertgroupId, userIds);
- return returnDataList(result);
- }
}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java
index 20f4285..b908b05 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java
@@ -14,30 +14,48 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.api.controller;
+import static org.apache.dolphinscheduler.api.enums.Status.CHECK_PROCESS_DEFINITION_ERROR;
+import static org.apache.dolphinscheduler.api.enums.Status.EXECUTE_PROCESS_INSTANCE_ERROR;
+import static org.apache.dolphinscheduler.api.enums.Status.START_PROCESS_INSTANCE_ERROR;
import org.apache.dolphinscheduler.api.enums.ExecuteType;
import org.apache.dolphinscheduler.api.exceptions.ApiException;
import org.apache.dolphinscheduler.api.service.ExecutorService;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.enums.CommandType;
+import org.apache.dolphinscheduler.common.enums.FailureStrategy;
+import org.apache.dolphinscheduler.common.enums.Priority;
+import org.apache.dolphinscheduler.common.enums.RunMode;
+import org.apache.dolphinscheduler.common.enums.TaskDependType;
+import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.dao.entity.User;
-import io.swagger.annotations.*;
-import org.apache.dolphinscheduler.common.enums.*;
+
+import java.text.ParseException;
+import java.util.Map;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
-import org.springframework.web.bind.annotation.*;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestAttribute;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseStatus;
+import org.springframework.web.bind.annotation.RestController;
+
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiImplicitParam;
+import io.swagger.annotations.ApiImplicitParams;
+import io.swagger.annotations.ApiOperation;
+import io.swagger.annotations.ApiParam;
import springfox.documentation.annotations.ApiIgnore;
-import java.text.ParseException;
-import java.util.Map;
-
-import static org.apache.dolphinscheduler.api.enums.Status.*;
-
-
/**
* execute process controller
*/
@@ -64,8 +82,6 @@ public class ExecutorController extends BaseController {
* @param execType execute type
* @param warningType warning type
* @param warningGroupId warning group id
- * @param receivers receivers
- * @param receiversCc receivers cc
* @param runMode run mode
* @param processInstancePriority process instance priority
* @param workerGroup worker group
@@ -74,20 +90,18 @@ public class ExecutorController extends BaseController {
*/
@ApiOperation(value = "startProcessInstance", notes = "RUN_PROCESS_INSTANCE_NOTES")
@ApiImplicitParams({
- @ApiImplicitParam(name = "processDefinitionId", value = "PROCESS_DEFINITION_ID", required = true, dataType = "Int", example = "100"),
- @ApiImplicitParam(name = "scheduleTime", value = "SCHEDULE_TIME", required = true, dataType = "String"),
- @ApiImplicitParam(name = "failureStrategy", value = "FAILURE_STRATEGY", required = true, dataType = "FailureStrategy"),
- @ApiImplicitParam(name = "startNodeList", value = "START_NODE_LIST", dataType = "String"),
- @ApiImplicitParam(name = "taskDependType", value = "TASK_DEPEND_TYPE", dataType = "TaskDependType"),
- @ApiImplicitParam(name = "execType", value = "COMMAND_TYPE", dataType = "CommandType"),
- @ApiImplicitParam(name = "warningType", value = "WARNING_TYPE", required = true, dataType = "WarningType"),
- @ApiImplicitParam(name = "warningGroupId", value = "WARNING_GROUP_ID", required = true, dataType = "Int", example = "100"),
- @ApiImplicitParam(name = "receivers", value = "RECEIVERS", dataType = "String"),
- @ApiImplicitParam(name = "receiversCc", value = "RECEIVERS_CC", dataType = "String"),
- @ApiImplicitParam(name = "runMode", value = "RUN_MODE", dataType = "RunMode"),
- @ApiImplicitParam(name = "processInstancePriority", value = "PROCESS_INSTANCE_PRIORITY", required = true, dataType = "Priority"),
- @ApiImplicitParam(name = "workerGroup", value = "WORKER_GROUP", dataType = "String", example = "default"),
- @ApiImplicitParam(name = "timeout", value = "TIMEOUT", dataType = "Int", example = "100"),
+ @ApiImplicitParam(name = "processDefinitionId", value = "PROCESS_DEFINITION_ID", required = true, dataType = "Int", example = "100"),
+ @ApiImplicitParam(name = "scheduleTime", value = "SCHEDULE_TIME", required = true, dataType = "String"),
+ @ApiImplicitParam(name = "failureStrategy", value = "FAILURE_STRATEGY", required = true, dataType = "FailureStrategy"),
+ @ApiImplicitParam(name = "startNodeList", value = "START_NODE_LIST", dataType = "String"),
+ @ApiImplicitParam(name = "taskDependType", value = "TASK_DEPEND_TYPE", dataType = "TaskDependType"),
+ @ApiImplicitParam(name = "execType", value = "COMMAND_TYPE", dataType = "CommandType"),
+ @ApiImplicitParam(name = "warningType", value = "WARNING_TYPE", required = true, dataType = "WarningType"),
+ @ApiImplicitParam(name = "warningGroupId", value = "WARNING_GROUP_ID", required = true, dataType = "Int", example = "100"),
+ @ApiImplicitParam(name = "runMode", value = "RUN_MODE", dataType = "RunMode"),
+ @ApiImplicitParam(name = "processInstancePriority", value = "PROCESS_INSTANCE_PRIORITY", required = true, dataType = "Priority"),
+ @ApiImplicitParam(name = "workerGroup", value = "WORKER_GROUP", dataType = "String", example = "default"),
+ @ApiImplicitParam(name = "timeout", value = "TIMEOUT", dataType = "Int", example = "100"),
})
@PostMapping(value = "start-process-instance")
@ResponseStatus(HttpStatus.OK)
@@ -102,26 +116,24 @@ public class ExecutorController extends BaseController {
@RequestParam(value = "execType", required = false) CommandType execType,
@RequestParam(value = "warningType", required = true) WarningType warningType,
@RequestParam(value = "warningGroupId", required = false) int warningGroupId,
- @RequestParam(value = "receivers", required = false) String receivers,
- @RequestParam(value = "receiversCc", required = false) String receiversCc,
@RequestParam(value = "runMode", required = false) RunMode runMode,
@RequestParam(value = "processInstancePriority", required = false) Priority processInstancePriority,
@RequestParam(value = "workerGroup", required = false, defaultValue = "default") String workerGroup,
@RequestParam(value = "timeout", required = false) Integer timeout) throws ParseException {
logger.info("login user {}, start process instance, project name: {}, process definition id: {}, schedule time: {}, "
- + "failure policy: {}, node name: {}, node dep: {}, notify type: {}, "
- + "notify group id: {},receivers:{},receiversCc:{}, run mode: {},process instance priority:{}, workerGroup: {}, timeout: {}",
- loginUser.getUserName(), projectName, processDefinitionId, scheduleTime,
- failureStrategy, startNodeList, taskDependType, warningType, workerGroup, receivers, receiversCc, runMode, processInstancePriority,
- workerGroup, timeout);
+ + "failure policy: {}, node name: {}, node dep: {}, notify type: {}, "
+ + "notify group id: {}, run mode: {},process instance priority:{}, workerGroup: {}, timeout: {}",
+ loginUser.getUserName(), projectName, processDefinitionId, scheduleTime,
+ failureStrategy, startNodeList, taskDependType, warningType, workerGroup, runMode, processInstancePriority,
+ workerGroup, timeout);
if (timeout == null) {
timeout = Constants.MAX_TASK_TIMEOUT;
}
Map<String, Object> result = execService.execProcessInstance(loginUser, projectName, processDefinitionId, scheduleTime, execType, failureStrategy,
- startNodeList, taskDependType, warningType,
- warningGroupId, receivers, receiversCc, runMode, processInstancePriority, workerGroup, timeout);
+ startNodeList, taskDependType, warningType,
+ warningGroupId, runMode, processInstancePriority, workerGroup, timeout);
return returnDataList(result);
}
@@ -137,8 +149,8 @@ public class ExecutorController extends BaseController {
*/
@ApiOperation(value = "execute", notes = "EXECUTE_ACTION_TO_PROCESS_INSTANCE_NOTES")
@ApiImplicitParams({
- @ApiImplicitParam(name = "processInstanceId", value = "PROCESS_INSTANCE_ID", required = true, dataType = "Int", example = "100"),
- @ApiImplicitParam(name = "executeType", value = "EXECUTE_TYPE", required = true, dataType = "ExecuteType")
+ @ApiImplicitParam(name = "processInstanceId", value = "PROCESS_INSTANCE_ID", required = true, dataType = "Int", example = "100"),
+ @ApiImplicitParam(name = "executeType", value = "EXECUTE_TYPE", required = true, dataType = "ExecuteType")
})
@PostMapping(value = "/execute")
@ResponseStatus(HttpStatus.OK)
@@ -149,7 +161,7 @@ public class ExecutorController extends BaseController {
@RequestParam("executeType") ExecuteType executeType
) {
logger.info("execute command, login user: {}, project:{}, process instance id:{}, execute type:{}",
- loginUser.getUserName(), projectName, processInstanceId, executeType);
+ loginUser.getUserName(), projectName, processInstanceId, executeType);
Map<String, Object> result = execService.execute(loginUser, projectName, processInstanceId, executeType);
return returnDataList(result);
}
@@ -163,7 +175,7 @@ public class ExecutorController extends BaseController {
*/
@ApiOperation(value = "startCheckProcessDefinition", notes = "START_CHECK_PROCESS_DEFINITION_NOTES")
@ApiImplicitParams({
- @ApiImplicitParam(name = "processDefinitionId", value = "PROCESS_DEFINITION_ID", required = true, dataType = "Int", example = "100")
+ @ApiImplicitParam(name = "processDefinitionId", value = "PROCESS_DEFINITION_ID", required = true, dataType = "Int", example = "100")
})
@PostMapping(value = "/start-check")
@ResponseStatus(HttpStatus.OK)
@@ -174,32 +186,4 @@ public class ExecutorController extends BaseController {
Map<String, Object> result = execService.startCheckByProcessDefinedId(processDefinitionId);
return returnDataList(result);
}
-
- /**
- * query recipients and copyers by process definition ID
- *
- * @param loginUser login user
- * @param processDefinitionId process definition id
- * @param processInstanceId process instance id
- * @return receivers cc list
- */
- @ApiIgnore
- @ApiOperation(value = "getReceiverCc", notes = "GET_RECEIVER_CC_NOTES")
- @ApiImplicitParams({
- @ApiImplicitParam(name = "processDefinitionId", value = "PROCESS_DEFINITION_ID", required = true, dataType = "Int", example = "100"),
- @ApiImplicitParam(name = "processInstanceId", value = "PROCESS_INSTANCE_ID", required = true, dataType = "Int", example = "100")
-
- })
- @GetMapping(value = "/get-receiver-cc")
- @ResponseStatus(HttpStatus.OK)
- @ApiException(QUERY_RECIPIENTS_AND_COPYERS_BY_PROCESS_DEFINITION_ERROR)
- public Result getReceiverCc(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
- @RequestParam(value = "processDefinitionId", required = false) Integer processDefinitionId,
- @RequestParam(value = "processInstanceId", required = false) Integer processInstanceId) {
- logger.info("login user {}, get process definition receiver and cc", loginUser.getUserName());
- Map<String, Object> result = execService.getReceiverCc(processDefinitionId, processInstanceId);
- return returnDataList(result);
- }
-
-
}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/AlertGroupService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/AlertGroupService.java
index 72cbd50..d4602d3 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/AlertGroupService.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/AlertGroupService.java
@@ -14,9 +14,9 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.api.service;
-import java.util.*;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.common.Constants;
@@ -25,29 +25,33 @@ import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.StringUtils;
import org.apache.dolphinscheduler.dao.entity.AlertGroup;
import org.apache.dolphinscheduler.dao.entity.User;
-import org.apache.dolphinscheduler.dao.entity.UserAlertGroup;
import org.apache.dolphinscheduler.dao.mapper.AlertGroupMapper;
-import com.baomidou.mybatisplus.core.metadata.IPage;
-import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+
/**
* alert group service
*/
@Service
-public class AlertGroupService extends BaseService{
+public class AlertGroupService extends BaseService {
private static final Logger logger = LoggerFactory.getLogger(AlertGroupService.class);
@Autowired
private AlertGroupMapper alertGroupMapper;
- @Autowired
- private UserAlertGroupService userAlertGroupService;
/**
* query alert group list
*
@@ -68,8 +72,8 @@ public class AlertGroupService extends BaseService{
*
* @param loginUser login user
* @param searchVal search value
- * @param pageNo page number
- * @param pageSize page size
+ * @param pageNo page number
+ * @param pageSize page size
* @return alert group list page
*/
public Map<String, Object> listPaging(User loginUser, String searchVal, Integer pageNo, Integer pageSize) {
@@ -81,9 +85,9 @@ public class AlertGroupService extends BaseService{
Page<AlertGroup> page = new Page(pageNo, pageSize);
IPage<AlertGroup> alertGroupIPage = alertGroupMapper.queryAlertGroupPage(
- page, searchVal);
+ page, searchVal);
PageInfo<AlertGroup> pageInfo = new PageInfo<>(pageNo, pageSize);
- pageInfo.setTotalCount((int)alertGroupIPage.getTotal());
+ pageInfo.setTotalCount((int) alertGroupIPage.getTotal());
pageInfo.setLists(alertGroupIPage.getRecords());
result.put(Constants.DATA_LIST, pageInfo);
putMsg(result, Status.SUCCESS);
@@ -97,13 +101,13 @@ public class AlertGroupService extends BaseService{
* @param loginUser login user
* @param groupName group name
* @param groupType group type
- * @param desc description
+ * @param desc description
* @return create result code
*/
public Map<String, Object> createAlertgroup(User loginUser, String groupName, AlertType groupType, String desc) {
Map<String, Object> result = new HashMap<>();
//only admin can operate
- if (checkAdmin(loginUser, result)){
+ if (checkAdmin(loginUser, result)) {
return result;
}
@@ -111,10 +115,10 @@ public class AlertGroupService extends BaseService{
Date now = new Date();
alertGroup.setGroupName(groupName);
- alertGroup.setGroupType(groupType);
alertGroup.setDescription(desc);
alertGroup.setCreateTime(now);
alertGroup.setUpdateTime(now);
+ alertGroup.setCreateUserId(loginUser.getId());
// insert
int insert = alertGroupMapper.insert(alertGroup);
@@ -131,20 +135,19 @@ public class AlertGroupService extends BaseService{
* updateProcessInstance alert group
*
* @param loginUser login user
- * @param id alert group id
+ * @param id alert group id
* @param groupName group name
* @param groupType group type
- * @param desc description
+ * @param desc description
* @return update result code
*/
public Map<String, Object> updateAlertgroup(User loginUser, int id, String groupName, AlertType groupType, String desc) {
Map<String, Object> result = new HashMap<>();
- if (checkAdmin(loginUser, result)){
+ if (checkAdmin(loginUser, result)) {
return result;
}
-
AlertGroup alertGroup = alertGroupMapper.selectById(id);
if (alertGroup == null) {
@@ -158,12 +161,9 @@ public class AlertGroupService extends BaseService{
if (StringUtils.isNotEmpty(groupName)) {
alertGroup.setGroupName(groupName);
}
-
- if (groupType != null) {
- alertGroup.setGroupType(groupType);
- }
alertGroup.setDescription(desc);
alertGroup.setUpdateTime(now);
+ alertGroup.setCreateUserId(loginUser.getId());
// updateProcessInstance
alertGroupMapper.updateById(alertGroup);
putMsg(result, Status.SUCCESS);
@@ -174,7 +174,7 @@ public class AlertGroupService extends BaseService{
* delete alert group by id
*
* @param loginUser login user
- * @param id alert group id
+ * @param id alert group id
* @return delete result code
*/
@Transactional(rollbackFor = RuntimeException.class)
@@ -183,7 +183,7 @@ public class AlertGroupService extends BaseService{
result.put(Constants.STATUS, false);
//only admin can operate
- if (checkAdmin(loginUser, result)){
+ if (checkAdmin(loginUser, result)) {
return result;
}
//check exist
@@ -192,57 +192,11 @@ public class AlertGroupService extends BaseService{
putMsg(result, Status.ALERT_GROUP_NOT_EXIST);
return result;
}
-
- userAlertGroupService.deleteByAlertGroupId(id);
alertGroupMapper.deleteById(id);
putMsg(result, Status.SUCCESS);
return result;
}
-
- /**
- * grant user
- *
- * @param loginUser login user
- * @param alertgroupId alert group id
- * @param userIds user id list
- * @return grant result code
- */
- public Map<String, Object> grantUser(User loginUser, int alertgroupId, String userIds) {
- Map<String, Object> result = new HashMap<>();
- result.put(Constants.STATUS, false);
-
- //only admin can operate
- if (checkAdmin(loginUser, result)){
- return result;
- }
-
- userAlertGroupService.deleteByAlertGroupId(alertgroupId);
- if (StringUtils.isEmpty(userIds)) {
- putMsg(result, Status.SUCCESS);
- return result;
- }
-
- String[] userIdsArr = userIds.split(",");
- Date now = new Date();
- List<UserAlertGroup> alertGroups = new ArrayList<>(userIds.length());
- for (String userId : userIdsArr) {
- UserAlertGroup userAlertGroup = new UserAlertGroup();
- userAlertGroup.setAlertgroupId(alertgroupId);
- userAlertGroup.setUserId(Integer.parseInt(userId));
- userAlertGroup.setCreateTime(now);
- userAlertGroup.setUpdateTime(now);
- alertGroups.add(userAlertGroup);
- }
-
- if (CollectionUtils.isNotEmpty(alertGroups)) {
- userAlertGroupService.saveBatch(alertGroups);
- }
-
- putMsg(result, Status.SUCCESS);
- return result;
- }
-
/**
* verify group name exists
*
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java
index fb735ec..e5e1513 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java
@@ -83,8 +83,6 @@ public class ExecutorService extends BaseService{
* @param taskDependType node dependency type
* @param warningType warning type
* @param warningGroupId notify group id
- * @param receivers receivers
- * @param receiversCc receivers cc
* @param processInstancePriority process instance priority
* @param workerGroup worker group name
* @param runMode run mode
@@ -96,7 +94,7 @@ public class ExecutorService extends BaseService{
int processDefinitionId, String cronTime, CommandType commandType,
FailureStrategy failureStrategy, String startNodeList,
TaskDependType taskDependType, WarningType warningType, int warningGroupId,
- String receivers, String receiversCc, RunMode runMode,
+ RunMode runMode,
Priority processInstancePriority, String workerGroup, Integer timeout) throws ParseException {
Map<String, Object> result = new HashMap<>();
// timeout is invalid
@@ -137,11 +135,7 @@ public class ExecutorService extends BaseService{
taskDependType, failureStrategy, startNodeList, cronTime, warningType, loginUser.getId(),
warningGroupId, runMode,processInstancePriority, workerGroup);
if(create > 0 ){
- /**
- * according to the process definition ID updateProcessInstance and CC recipient
- */
- processDefinition.setReceivers(receivers);
- processDefinition.setReceiversCc(receiversCc);
+ processDefinition.setWarningGroupId(warningGroupId);
processDefinitionMapper.updateById(processDefinition);
putMsg(result, Status.SUCCESS);
} else {
@@ -425,43 +419,6 @@ public class ExecutorService extends BaseService{
}
/**
- * query recipients and copyers by process definition id or processInstanceId
- *
- * @param processDefineId process definition id
- * @param processInstanceId process instance id
- * @return receivers cc list
- */
- public Map<String, Object> getReceiverCc(Integer processDefineId,Integer processInstanceId) {
- Map<String, Object> result = new HashMap<>();
- logger.info("processInstanceId {}",processInstanceId);
- if(processDefineId == null && processInstanceId == null){
- throw new RuntimeException("You must set values for parameters processDefineId or processInstanceId");
- }
- if(processDefineId == null && processInstanceId != null) {
- ProcessInstance processInstance = processInstanceMapper.selectById(processInstanceId);
- if (processInstance == null) {
- throw new RuntimeException("processInstanceId is not exists");
- }
- processDefineId = processInstance.getProcessDefinitionId();
- }
- ProcessDefinition processDefinition = processDefinitionMapper.selectById(processDefineId);
- if (processDefinition == null){
- throw new RuntimeException(String.format("processDefineId %d is not exists",processDefineId));
- }
-
- String receivers = processDefinition.getReceivers();
- String receiversCc = processDefinition.getReceiversCc();
- Map<String,String> dataMap = new HashMap<>();
- dataMap.put(Constants.RECEIVERS,receivers);
- dataMap.put(Constants.RECEIVERS_CC,receiversCc);
-
- result.put(Constants.DATA_LIST, dataMap);
- putMsg(result, Status.SUCCESS);
- return result;
- }
-
-
- /**
* create command
* @param commandType commandType
* @param processDefineId processDefineId
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessInstanceService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessInstanceService.java
index c8d3c74..24fa871 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessInstanceService.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessInstanceService.java
@@ -187,8 +187,7 @@ public class ProcessInstanceService extends BaseService {
ProcessInstance processInstance = processService.findProcessInstanceDetailById(processId);
ProcessDefinition processDefinition = processService.findProcessDefineById(processInstance.getProcessDefinitionId());
- processInstance.setReceivers(processDefinition.getReceivers());
- processInstance.setReceiversCc(processDefinition.getReceiversCc());
+ processInstance.setWarningGroupId(processDefinition.getWarningGroupId());
result.put(DATA_LIST, processInstance);
putMsg(result, Status.SUCCESS);
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/SchedulerService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/SchedulerService.java
index 93fa148..09ad58c 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/SchedulerService.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/SchedulerService.java
@@ -161,8 +161,7 @@ public class SchedulerService extends BaseService {
/**
* updateProcessInstance receivers and cc by process definition id
*/
- processDefinition.setReceivers(receivers);
- processDefinition.setReceiversCc(receiversCc);
+ processDefinition.setWarningGroupId(warningGroupId);
processDefinitionMapper.updateById(processDefinition);
// return scheduler object with ID
@@ -275,8 +274,8 @@ public class SchedulerService extends BaseService {
/**
* updateProcessInstance recipients and cc by process definition ID
*/
- processDefinition.setReceivers(receivers);
- processDefinition.setReceiversCc(receiversCc);
+ processDefinition.setWarningGroupId(warningGroupId);
+
processDefinitionMapper.updateById(processDefinition);
putMsg(result, Status.SUCCESS);
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/UserAlertGroupService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/UserAlertGroupService.java
deleted file mode 100644
index 5021857..0000000
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/UserAlertGroupService.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.dolphinscheduler.api.service;
-
-import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
-import org.apache.dolphinscheduler.dao.entity.UserAlertGroup;
-import org.apache.dolphinscheduler.dao.mapper.UserAlertGroupMapper;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
-/**
- *
- */
-@Service
-public class UserAlertGroupService extends ServiceImpl<UserAlertGroupMapper, UserAlertGroup> {
-
- @Autowired
- private UserAlertGroupMapper userAlertGroupMapper;
-
- boolean deleteByAlertGroupId(Integer groupId) {
- return userAlertGroupMapper.deleteByAlertgroupId(groupId) >= 1;
- }
-
-}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/UsersService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/UsersService.java
index 89038ad..92137be 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/UsersService.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/UsersService.java
@@ -14,10 +14,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.api.service;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+
import org.apache.dolphinscheduler.api.dto.resources.ResourceComponent;
import org.apache.dolphinscheduler.api.dto.resources.visitor.ResourceTreeVisitor;
import org.apache.dolphinscheduler.api.enums.Status;
@@ -33,6 +35,7 @@ import org.apache.dolphinscheduler.common.utils.*;
import org.apache.dolphinscheduler.dao.entity.*;
import org.apache.dolphinscheduler.dao.mapper.*;
import org.apache.dolphinscheduler.dao.utils.ResourceProcessDefinitionUtils;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -83,13 +86,13 @@ public class UsersService extends BaseService {
/**
* create user, only system admin have permission
*
- * @param loginUser login user
- * @param userName user name
+ * @param loginUser login user
+ * @param userName user name
* @param userPassword user password
- * @param email email
- * @param tenantId tenant id
- * @param phone phone
- * @param queue queue
+ * @param email email
+ * @param tenantId tenant id
+ * @param phone phone
+ * @param queue queue
* @return create result code
* @throws Exception exception
*/
@@ -109,7 +112,7 @@ public class UsersService extends BaseService {
String msg = this.checkUserParams(userName, userPassword, email, phone);
if (!StringUtils.isEmpty(msg)) {
- putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR,msg);
+ putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, msg);
return result;
}
if (!isAdmin(loginUser)) {
@@ -126,12 +129,12 @@ public class UsersService extends BaseService {
Tenant tenant = tenantMapper.queryById(tenantId);
// resource upload startup
- if (PropertyUtils.getResUploadStartupState()){
+ if (PropertyUtils.getResUploadStartupState()) {
// if tenant not exists
- if (!HadoopUtils.getInstance().exists(HadoopUtils.getHdfsTenantDir(tenant.getTenantCode()))){
+ if (!HadoopUtils.getInstance().exists(HadoopUtils.getHdfsTenantDir(tenant.getTenantCode()))) {
createTenantDirIfNotExists(tenant.getTenantCode());
}
- String userPath = HadoopUtils.getHdfsUserDir(tenant.getTenantCode(),user.getId());
+ String userPath = HadoopUtils.getHdfsUserDir(tenant.getTenantCode(), user.getId());
HadoopUtils.getInstance().mkdir(userPath);
}
@@ -142,12 +145,12 @@ public class UsersService extends BaseService {
@Transactional(rollbackFor = RuntimeException.class)
public User createUser(String userName,
- String userPassword,
- String email,
- int tenantId,
- String phone,
- String queue,
- int state) {
+ String userPassword,
+ String email,
+ int tenantId,
+ String phone,
+ String queue,
+ int state) {
User user = new User();
Date now = new Date();
@@ -161,7 +164,7 @@ public class UsersService extends BaseService {
user.setUserType(UserType.GENERAL_USER);
user.setCreateTime(now);
user.setUpdateTime(now);
- if (StringUtils.isEmpty(queue)){
+ if (StringUtils.isEmpty(queue)) {
queue = "";
}
user.setQueue(queue);
@@ -173,6 +176,7 @@ public class UsersService extends BaseService {
/**
* query user by id
+ *
* @param id id
* @return user info
*/
@@ -182,6 +186,7 @@ public class UsersService extends BaseService {
/**
* query user
+ *
* @param name name
* @return user info
*/
@@ -192,7 +197,7 @@ public class UsersService extends BaseService {
/**
* query user
*
- * @param name name
+ * @param name name
* @param password password
* @return user info
*/
@@ -203,6 +208,7 @@ public class UsersService extends BaseService {
/**
* get user id by user name
+ *
* @param name user name
* @return if name empty 0, user not exists -1, user exist user id
*/
@@ -225,9 +231,9 @@ public class UsersService extends BaseService {
* query user list
*
* @param loginUser login user
- * @param pageNo page number
+ * @param pageNo page number
* @param searchVal search avlue
- * @param pageSize page size
+ * @param pageSize page size
* @return user list page
*/
public Map<String, Object> queryUserList(User loginUser, String searchVal, Integer pageNo, Integer pageSize) {
@@ -242,7 +248,7 @@ public class UsersService extends BaseService {
IPage<User> scheduleList = userMapper.queryUserPaging(page, searchVal);
PageInfo<User> pageInfo = new PageInfo<>(pageNo, pageSize);
- pageInfo.setTotalCount((int)scheduleList.getTotal());
+ pageInfo.setTotalCount((int) scheduleList.getTotal());
pageInfo.setLists(scheduleList.getRecords());
result.put(Constants.DATA_LIST, pageInfo);
putMsg(result, Status.SUCCESS);
@@ -253,13 +259,13 @@ public class UsersService extends BaseService {
/**
* updateProcessInstance user
*
- * @param userId user id
- * @param userName user name
+ * @param userId user id
+ * @param userName user name
* @param userPassword user password
- * @param email email
- * @param tenantId tennat id
- * @param phone phone
- * @param queue queue
+ * @param email email
+ * @param tenantId tennat id
+ * @param phone phone
+ * @param queue queue
* @return update result code
* @throws Exception exception
*/
@@ -283,8 +289,8 @@ public class UsersService extends BaseService {
if (StringUtils.isNotEmpty(userName)) {
- if (!CheckUtils.checkUserName(userName)){
- putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR,userName);
+ if (!CheckUtils.checkUserName(userName)) {
+ putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, userName);
return result;
}
@@ -297,23 +303,23 @@ public class UsersService extends BaseService {
}
if (StringUtils.isNotEmpty(userPassword)) {
- if (!CheckUtils.checkPassword(userPassword)){
- putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR,userPassword);
+ if (!CheckUtils.checkPassword(userPassword)) {
+ putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, userPassword);
return result;
}
user.setUserPassword(EncryptionUtils.getMd5(userPassword));
}
if (StringUtils.isNotEmpty(email)) {
- if (!CheckUtils.checkEmail(email)){
- putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR,email);
+ if (!CheckUtils.checkEmail(email)) {
+ putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, email);
return result;
}
user.setEmail(email);
}
if (StringUtils.isNotEmpty(phone) && !CheckUtils.checkPhone(phone)) {
- putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR,phone);
+ putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, phone);
return result;
}
user.setPhone(phone);
@@ -329,19 +335,19 @@ public class UsersService extends BaseService {
Tenant newTenant = tenantMapper.queryById(tenantId);
if (newTenant != null) {
// if hdfs startup
- if (PropertyUtils.getResUploadStartupState() && oldTenant != null){
+ if (PropertyUtils.getResUploadStartupState() && oldTenant != null) {
String newTenantCode = newTenant.getTenantCode();
String oldResourcePath = HadoopUtils.getHdfsResDir(oldTenant.getTenantCode());
String oldUdfsPath = HadoopUtils.getHdfsUdfDir(oldTenant.getTenantCode());
// if old tenant dir exists
- if (HadoopUtils.getInstance().exists(oldResourcePath)){
+ if (HadoopUtils.getInstance().exists(oldResourcePath)) {
String newResourcePath = HadoopUtils.getHdfsResDir(newTenantCode);
String newUdfsPath = HadoopUtils.getHdfsUdfDir(newTenantCode);
//file resources list
List<Resource> fileResourcesList = resourceMapper.queryResourceList(
- null, userId, ResourceType.FILE.ordinal());
+ null, userId, ResourceType.FILE.ordinal());
if (CollectionUtils.isNotEmpty(fileResourcesList)) {
ResourceTreeVisitor resourceTreeVisitor = new ResourceTreeVisitor(fileResourcesList);
ResourceComponent resourceComponent = resourceTreeVisitor.visit();
@@ -350,7 +356,7 @@ public class UsersService extends BaseService {
//udf resources
List<Resource> udfResourceList = resourceMapper.queryResourceList(
- null, userId, ResourceType.UDF.ordinal());
+ null, userId, ResourceType.UDF.ordinal());
if (CollectionUtils.isNotEmpty(udfResourceList)) {
ResourceTreeVisitor resourceTreeVisitor = new ResourceTreeVisitor(udfResourceList);
ResourceComponent resourceComponent = resourceTreeVisitor.visit();
@@ -358,18 +364,18 @@ public class UsersService extends BaseService {
}
//Delete the user from the old tenant directory
- String oldUserPath = HadoopUtils.getHdfsUserDir(oldTenant.getTenantCode(),userId);
+ String oldUserPath = HadoopUtils.getHdfsUserDir(oldTenant.getTenantCode(), userId);
HadoopUtils.getInstance().delete(oldUserPath, true);
- }else {
+ } else {
// if old tenant dir not exists , create
createTenantDirIfNotExists(oldTenant.getTenantCode());
}
- if (HadoopUtils.getInstance().exists(HadoopUtils.getHdfsTenantDir(newTenant.getTenantCode()))){
+ if (HadoopUtils.getInstance().exists(HadoopUtils.getHdfsTenantDir(newTenant.getTenantCode()))) {
//create user in the new tenant directory
- String newUserPath = HadoopUtils.getHdfsUserDir(newTenant.getTenantCode(),user.getId());
+ String newUserPath = HadoopUtils.getHdfsUserDir(newTenant.getTenantCode(), user.getId());
HadoopUtils.getInstance().mkdir(newUserPath);
- }else {
+ } else {
// if new tenant dir not exists , create
createTenantDirIfNotExists(newTenant.getTenantCode());
}
@@ -389,7 +395,7 @@ public class UsersService extends BaseService {
* delete user
*
* @param loginUser login user
- * @param id user id
+ * @param id user id
* @return delete result code
* @throws Exception exception when operate hdfs
*/
@@ -411,7 +417,7 @@ public class UsersService extends BaseService {
if (user != null) {
if (PropertyUtils.getResUploadStartupState()) {
- String userPath = HadoopUtils.getHdfsUserDir(user.getTenantCode(),id);
+ String userPath = HadoopUtils.getHdfsUserDir(user.getTenantCode(), id);
if (HadoopUtils.getInstance().exists(userPath)) {
HadoopUtils.getInstance().delete(userPath, true);
}
@@ -427,8 +433,8 @@ public class UsersService extends BaseService {
/**
* grant project
*
- * @param loginUser login user
- * @param userId user id
+ * @param loginUser login user
+ * @param userId user id
* @param projectIds project id array
* @return grant result code
*/
@@ -477,8 +483,8 @@ public class UsersService extends BaseService {
/**
* grant resource
*
- * @param loginUser login user
- * @param userId user id
+ * @param loginUser login user
+ * @param userId user id
* @param resourceIds resource id array
* @return grant result code
*/
@@ -490,7 +496,7 @@ public class UsersService extends BaseService {
return result;
}
User user = userMapper.selectById(userId);
- if(user == null){
+ if (user == null) {
putMsg(result, Status.USER_NOT_EXIST, userId);
return result;
}
@@ -501,7 +507,7 @@ public class UsersService extends BaseService {
// need authorize resource id set
for (String resourceFullId : resourceFullIdArr) {
String[] resourceIdArr = resourceFullId.split("-");
- for (int i=0;i<=resourceIdArr.length-1;i++) {
+ for (int i = 0; i <= resourceIdArr.length - 1; i++) {
int resourceIdValue = Integer.parseInt(resourceIdArr[i]);
needAuthorizeResIds.add(resourceIdValue);
}
@@ -528,7 +534,7 @@ public class UsersService extends BaseService {
if (CollectionUtils.isNotEmpty(resourceIdSet)) {
logger.error("can't be deleted,because it is used of process definition");
for (Integer resId : resourceIdSet) {
- logger.error("resource id:{} is used of process definition {}",resId,resourceProcessMap.get(resId));
+ logger.error("resource id:{} is used of process definition {}", resId, resourceProcessMap.get(resId));
}
putMsg(result, Status.RESOURCE_IS_USED);
return result;
@@ -555,7 +561,7 @@ public class UsersService extends BaseService {
resourcesUser.setResourcesId(resourceIdValue);
if (resource.isDirectory()) {
resourcesUser.setPerm(Constants.AUTHORIZE_READABLE_PERM);
- }else{
+ } else {
resourcesUser.setPerm(Constants.AUTHORIZE_WRITABLE_PERM);
}
@@ -575,8 +581,8 @@ public class UsersService extends BaseService {
* grant udf function
*
* @param loginUser login user
- * @param userId user id
- * @param udfIds udf id array
+ * @param userId user id
+ * @param udfIds udf id array
* @return grant result code
*/
@Transactional(rollbackFor = RuntimeException.class)
@@ -588,7 +594,7 @@ public class UsersService extends BaseService {
return result;
}
User user = userMapper.selectById(userId);
- if(user == null){
+ if (user == null) {
putMsg(result, Status.USER_NOT_EXIST, userId);
return result;
}
@@ -621,9 +627,9 @@ public class UsersService extends BaseService {
/**
* grant datasource
*
- * @param loginUser login user
- * @param userId user id
- * @param datasourceIds data source id array
+ * @param loginUser login user
+ * @param userId user id
+ * @param datasourceIds data source id array
* @return grant result code
*/
@Transactional(rollbackFor = RuntimeException.class)
@@ -636,7 +642,7 @@ public class UsersService extends BaseService {
return result;
}
User user = userMapper.selectById(userId);
- if(user == null){
+ if (user == null) {
putMsg(result, Status.USER_NOT_EXIST, userId);
return result;
}
@@ -676,23 +682,11 @@ public class UsersService extends BaseService {
Map<String, Object> result = new HashMap<>();
- User user = null;
+ User user;
if (loginUser.getUserType() == UserType.ADMIN_USER) {
user = loginUser;
} else {
user = userMapper.queryDetailsById(loginUser.getId());
-
- List<AlertGroup> alertGroups = alertGroupMapper.queryByUserId(loginUser.getId());
-
- StringBuilder sb = new StringBuilder();
-
- if (alertGroups != null && alertGroups.size() > 0) {
- for (int i = 0; i < alertGroups.size() - 1; i++) {
- sb.append(alertGroups.get(i).getGroupName() + ",");
- }
- sb.append(alertGroups.get(alertGroups.size() - 1));
- user.setAlertGroup(sb.toString());
- }
}
result.put(Constants.DATA_LIST, user);
@@ -735,7 +729,7 @@ public class UsersService extends BaseService {
return result;
}
- List<User> userList = userMapper.selectList(null );
+ List<User> userList = userMapper.selectList(null);
result.put(Constants.DATA_LIST, userList);
putMsg(result, Status.SUCCESS);
@@ -767,7 +761,7 @@ public class UsersService extends BaseService {
/**
* unauthorized user
*
- * @param loginUser login user
+ * @param loginUser login user
* @param alertgroupId alert group id
* @return unauthorize result code
*/
@@ -779,7 +773,7 @@ public class UsersService extends BaseService {
return result;
}
- List<User> userList = userMapper.selectList(null );
+ List<User> userList = userMapper.selectList(null);
List<User> resultUsers = new ArrayList<>();
Set<User> userSet = null;
if (userList != null && userList.size() > 0) {
@@ -804,7 +798,7 @@ public class UsersService extends BaseService {
/**
* authorized user
*
- * @param loginUser login user
+ * @param loginUser login user
* @param alertgroupId alert group id
* @return authorized result code
*/
@@ -824,8 +818,8 @@ public class UsersService extends BaseService {
/**
* check
*
- * @param result result
- * @param bool bool
+ * @param result result
+ * @param bool bool
* @param userNoOperationPerm status
* @return check result
*/
@@ -848,7 +842,6 @@ public class UsersService extends BaseService {
}
/**
- *
* @param userName
* @param password
* @param email
@@ -877,35 +870,36 @@ public class UsersService extends BaseService {
/**
* copy resource files
+ *
* @param resourceComponent resource component
* @param srcBasePath src base path
* @param dstBasePath dst base path
- * @throws IOException io exception
+ * @throws IOException io exception
*/
private void copyResourceFiles(ResourceComponent resourceComponent, String srcBasePath, String dstBasePath) throws IOException {
List<ResourceComponent> components = resourceComponent.getChildren();
if (CollectionUtils.isNotEmpty(components)) {
- for (ResourceComponent component:components) {
+ for (ResourceComponent component : components) {
// verify whether exist
- if (!HadoopUtils.getInstance().exists(String.format("%s/%s",srcBasePath,component.getFullName()))){
- logger.error("resource file: {} not exist,copy error",component.getFullName());
+ if (!HadoopUtils.getInstance().exists(String.format("%s/%s", srcBasePath, component.getFullName()))) {
+ logger.error("resource file: {} not exist,copy error", component.getFullName());
throw new ServiceException(Status.RESOURCE_NOT_EXIST);
}
if (!component.isDirctory()) {
// copy it to dst
- HadoopUtils.getInstance().copy(String.format("%s/%s",srcBasePath,component.getFullName()),String.format("%s/%s",dstBasePath,component.getFullName()),false,true);
+ HadoopUtils.getInstance().copy(String.format("%s/%s", srcBasePath, component.getFullName()), String.format("%s/%s", dstBasePath, component.getFullName()), false, true);
continue;
}
- if(CollectionUtils.isEmpty(component.getChildren())) {
+ if (CollectionUtils.isEmpty(component.getChildren())) {
// if not exist,need create it
- if (!HadoopUtils.getInstance().exists(String.format("%s/%s",dstBasePath,component.getFullName()))) {
- HadoopUtils.getInstance().mkdir(String.format("%s/%s",dstBasePath,component.getFullName()));
+ if (!HadoopUtils.getInstance().exists(String.format("%s/%s", dstBasePath, component.getFullName()))) {
+ HadoopUtils.getInstance().mkdir(String.format("%s/%s", dstBasePath, component.getFullName()));
}
- }else{
- copyResourceFiles(component,srcBasePath,dstBasePath);
+ } else {
+ copyResourceFiles(component, srcBasePath, dstBasePath);
}
}
}
@@ -929,7 +923,7 @@ public class UsersService extends BaseService {
String msg = this.checkUserParams(userName, userPassword, email, "");
if (!StringUtils.isEmpty(msg)) {
- putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR,msg);
+ putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, msg);
return result;
}
@@ -959,7 +953,7 @@ public class UsersService extends BaseService {
return result;
}
- if (!CheckUtils.checkUserName(userName)){
+ if (!CheckUtils.checkUserName(userName)) {
putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, userName);
return result;
}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java
index 7b68003..9046f51 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java
@@ -158,7 +158,6 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
* @param locations locations for nodes
* @param connects connects for nodes
* @return create result code
- * @throws JsonProcessingException JsonProcessingException
*/
public Map<String, Object> createProcessDefinition(User loginUser,
String projectName,
@@ -166,7 +165,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
String processDefinitionJson,
String desc,
String locations,
- String connects) throws JsonProcessingException {
+ String connects) {
Map<String, Object> result = new HashMap<>();
Project project = projectMapper.queryByName(projectName);
@@ -836,19 +835,14 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
String processDefinitionName,
String importProcessParam) {
Map<String, Object> createProcessResult = null;
- try {
- createProcessResult = createProcessDefinition(loginUser
- , currentProjectName,
- processDefinitionName + "_import_" + DateUtils.getCurrentTimeStamp(),
- importProcessParam,
- processMeta.getProcessDefinitionDescription(),
- processMeta.getProcessDefinitionLocations(),
- processMeta.getProcessDefinitionConnects());
- putMsg(result, Status.SUCCESS);
- } catch (JsonProcessingException e) {
- logger.error("import process meta json data: {}", e.getMessage(), e);
- putMsg(result, Status.IMPORT_PROCESS_DEFINE_ERROR);
- }
+ createProcessResult = createProcessDefinition(loginUser
+ , currentProjectName,
+ processDefinitionName + "_import_" + DateUtils.getCurrentTimeStamp(),
+ importProcessParam,
+ processMeta.getProcessDefinitionDescription(),
+ processMeta.getProcessDefinitionLocations(),
+ processMeta.getProcessDefinitionConnects());
+ putMsg(result, Status.SUCCESS);
return createProcessResult;
}
@@ -1056,8 +1050,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
processDefine.setCreateTime(now);
processDefine.setUpdateTime(now);
processDefine.setFlag(subProcess.getFlag());
- processDefine.setReceivers(subProcess.getReceivers());
- processDefine.setReceiversCc(subProcess.getReceiversCc());
+ processDefine.setWarningGroupId(subProcess.getWarningGroupId());
processDefineMapper.insert(processDefine);
logger.info("create sub process, project: {}, process name: {}", targetProject.getName(), processDefine.getName());
@@ -1330,7 +1323,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
}
runningNodeMap.remove(nodeName);
}
- if (waitingRunningNodeMap == null || waitingRunningNodeMap.size() == 0) {
+ if (waitingRunningNodeMap.size() == 0) {
break;
} else {
runningNodeMap.putAll(waitingRunningNodeMap);
@@ -1580,8 +1573,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
processDefinition.setTimeout(processDefinitionVersion.getTimeout());
processDefinition.setGlobalParams(processDefinitionVersion.getGlobalParams());
processDefinition.setUpdateTime(new Date());
- processDefinition.setReceivers(processDefinitionVersion.getReceivers());
- processDefinition.setReceiversCc(processDefinitionVersion.getReceiversCc());
+ processDefinition.setWarningGroupId(processDefinitionVersion.getWarningGroupId());
processDefinition.setResourceIds(processDefinitionVersion.getResourceIds());
if (processDefineMapper.updateById(processDefinition) > 0) {
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionVersionServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionVersionServiceImpl.java
index 6364242..08cedfe 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionVersionServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionVersionServiceImpl.java
@@ -66,18 +66,17 @@ public class ProcessDefinitionVersionServiceImpl extends BaseService implements
long version = this.queryMaxVersionByProcessDefinitionId(processDefinition.getId()) + 1;
ProcessDefinitionVersion processDefinitionVersion = ProcessDefinitionVersion
- .newBuilder()
- .processDefinitionId(processDefinition.getId())
- .version(version)
- .processDefinitionJson(processDefinition.getProcessDefinitionJson())
- .description(processDefinition.getDescription())
- .locations(processDefinition.getLocations())
- .connects(processDefinition.getConnects())
- .timeout(processDefinition.getTimeout())
- .globalParams(processDefinition.getGlobalParams())
- .createTime(processDefinition.getUpdateTime())
- .receivers(processDefinition.getReceivers())
- .receiversCc(processDefinition.getReceiversCc())
+ .newBuilder()
+ .processDefinitionId(processDefinition.getId())
+ .version(version)
+ .processDefinitionJson(processDefinition.getProcessDefinitionJson())
+ .description(processDefinition.getDescription())
+ .locations(processDefinition.getLocations())
+ .connects(processDefinition.getConnects())
+ .timeout(processDefinition.getTimeout())
+ .globalParams(processDefinition.getGlobalParams())
+ .createTime(processDefinition.getUpdateTime())
+ .warningGroupId(processDefinition.getWarningGroupId())
.resourceIds(processDefinition.getResourceIds())
.build();
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/ExecutorControllerTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/ExecutorControllerTest.java
index 77d8f38..f0bd948 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/ExecutorControllerTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/ExecutorControllerTest.java
@@ -14,14 +14,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.api.controller;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
import org.apache.dolphinscheduler.api.enums.ExecuteType;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.enums.FailureStrategy;
import org.apache.dolphinscheduler.common.enums.WarningType;
-import org.apache.dolphinscheduler.common.utils.*;
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
+
import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Test;
@@ -32,44 +38,39 @@ import org.springframework.test.web.servlet.MvcResult;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
-import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
-import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
-import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
-import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
-
/**
* executor controller test
*/
-public class ExecutorControllerTest extends AbstractControllerTest{
+public class ExecutorControllerTest extends AbstractControllerTest {
private static Logger logger = LoggerFactory.getLogger(ExecutorControllerTest.class);
@Ignore
@Test
public void testStartProcessInstance() throws Exception {
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
- paramsMap.add("processDefinitionId","40");
- paramsMap.add("scheduleTime","");
+ paramsMap.add("processDefinitionId", "40");
+ paramsMap.add("scheduleTime", "");
paramsMap.add("failureStrategy", String.valueOf(FailureStrategy.CONTINUE));
- paramsMap.add("startNodeList","");
- paramsMap.add("taskDependType","");
- paramsMap.add("execType","");
+ paramsMap.add("startNodeList", "");
+ paramsMap.add("taskDependType", "");
+ paramsMap.add("execType", "");
paramsMap.add("warningType", String.valueOf(WarningType.NONE));
- paramsMap.add("warningGroupId","");
- paramsMap.add("receivers","");
- paramsMap.add("receiversCc","");
- paramsMap.add("runMode","");
- paramsMap.add("processInstancePriority","");
- paramsMap.add("workerGroupId","");
- paramsMap.add("timeout","");
+ paramsMap.add("warningGroupId", "");
+ paramsMap.add("receivers", "");
+ paramsMap.add("receiversCc", "");
+ paramsMap.add("runMode", "");
+ paramsMap.add("processInstancePriority", "");
+ paramsMap.add("workerGroupId", "");
+ paramsMap.add("timeout", "");
- MvcResult mvcResult = mockMvc.perform(post("/projects/{projectName}/executors/start-process-instance","cxc_1113")
- .header("sessionId", sessionId)
- .params(paramsMap))
- .andExpect(status().isOk())
- .andExpect(content().contentType(MediaType.APPLICATION_JSON_UTF8))
- .andReturn();
+ MvcResult mvcResult = mockMvc.perform(post("/projects/{projectName}/executors/start-process-instance", "cxc_1113")
+ .header("sessionId", sessionId)
+ .params(paramsMap))
+ .andExpect(status().isOk())
+ .andExpect(content().contentType(MediaType.APPLICATION_JSON_UTF8))
+ .andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
- Assert.assertEquals(Status.SUCCESS.getCode(),result.getCode().intValue());
+ Assert.assertEquals(Status.SUCCESS.getCode(), result.getCode().intValue());
logger.info(mvcResult.getResponse().getContentAsString());
}
@@ -77,47 +78,32 @@ public class ExecutorControllerTest extends AbstractControllerTest{
@Test
public void testExecute() throws Exception {
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
- paramsMap.add("processInstanceId","40");
- paramsMap.add("executeType",String.valueOf(ExecuteType.NONE));
+ paramsMap.add("processInstanceId", "40");
+ paramsMap.add("executeType", String.valueOf(ExecuteType.NONE));
- MvcResult mvcResult = mockMvc.perform(post("/projects/{projectName}/executors/execute","cxc_1113")
- .header("sessionId", sessionId)
- .params(paramsMap))
- .andExpect(status().isOk())
- .andExpect(content().contentType(MediaType.APPLICATION_JSON_UTF8))
- .andReturn();
+ MvcResult mvcResult = mockMvc.perform(post("/projects/{projectName}/executors/execute", "cxc_1113")
+ .header("sessionId", sessionId)
+ .params(paramsMap))
+ .andExpect(status().isOk())
+ .andExpect(content().contentType(MediaType.APPLICATION_JSON_UTF8))
+ .andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
- Assert.assertEquals(Status.SUCCESS.getCode(),result.getCode().intValue());
+ Assert.assertEquals(Status.SUCCESS.getCode(), result.getCode().intValue());
logger.info(mvcResult.getResponse().getContentAsString());
}
-
@Test
public void testStartCheckProcessDefinition() throws Exception {
- MvcResult mvcResult = mockMvc.perform(post("/projects/{projectName}/executors/start-check","cxc_1113")
- .header(SESSION_ID, sessionId)
- .param("processDefinitionId","40"))
- .andExpect(status().isOk())
- .andExpect(content().contentType(MediaType.APPLICATION_JSON_UTF8))
- .andReturn();
+ MvcResult mvcResult = mockMvc.perform(post("/projects/{projectName}/executors/start-check", "cxc_1113")
+ .header(SESSION_ID, sessionId)
+ .param("processDefinitionId", "40"))
+ .andExpect(status().isOk())
+ .andExpect(content().contentType(MediaType.APPLICATION_JSON_UTF8))
+ .andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
- Assert.assertEquals(Status.SUCCESS.getCode(),result.getCode().intValue());
+ Assert.assertEquals(Status.SUCCESS.getCode(), result.getCode().intValue());
logger.info(mvcResult.getResponse().getContentAsString());
}
- @Test
- public void testGetReceiverCc() throws Exception {
- MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
- paramsMap.add("processInstanceId","13");
- MvcResult mvcResult = mockMvc.perform(get("/projects/{projectName}/executors/get-receiver-cc","cxc_1113")
- .header(SESSION_ID, sessionId)
- .params(paramsMap))
- .andExpect(status().isOk())
- .andExpect(content().contentType(MediaType.APPLICATION_JSON_UTF8))
- .andReturn();
- Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
- Assert.assertEquals(Status.SUCCESS.getCode(),result.getCode().intValue());
- logger.info(mvcResult.getResponse().getContentAsString());
- }
}
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/AlertGroupServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/AlertGroupServiceTest.java
index 1b93e86..a4b7cb3 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/AlertGroupServiceTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/AlertGroupServiceTest.java
@@ -14,14 +14,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.api.service;
-import com.baomidou.mybatisplus.core.metadata.IPage;
-import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.common.Constants;
@@ -31,15 +29,15 @@ import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.dao.entity.AlertGroup;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.AlertGroupMapper;
-import org.apache.dolphinscheduler.dao.mapper.UserAlertGroupMapper;
-import org.junit.After;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
import org.junit.Assert;
-import static org.junit.Assert.assertEquals;
-import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
-import org.mockito.ArgumentCaptor;
-import static org.mockito.ArgumentMatchers.*;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
@@ -47,6 +45,9 @@ import org.mockito.junit.MockitoJUnitRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+
@RunWith(MockitoJUnitRunner.class)
public class AlertGroupServiceTest {
@@ -56,129 +57,98 @@ public class AlertGroupServiceTest {
private AlertGroupService alertGroupService;
@Mock
private AlertGroupMapper alertGroupMapper;
- @Mock
- private UserAlertGroupMapper userAlertGroupMapper;
- @Mock
- UserAlertGroupService userAlertGroupService;
private String groupName = "AlertGroupServiceTest";
- @Before
- public void setUp() {
- }
-
-
- @After
- public void after(){
-
- }
-
-
-
@Test
- public void testQueryAlertGroup(){
+ public void testQueryAlertGroup() {
Mockito.when(alertGroupMapper.queryAllGroupList()).thenReturn(getList());
- HashMap<String, Object> result= alertGroupService.queryAlertgroup();
+ HashMap<String, Object> result = alertGroupService.queryAlertgroup();
logger.info(result.toString());
List<AlertGroup> alertGroups = (List<AlertGroup>) result.get(Constants.DATA_LIST);
Assert.assertTrue(CollectionUtils.isNotEmpty(alertGroups));
}
+
@Test
- public void testListPaging(){
- IPage<AlertGroup> page = new Page<>(1,10);
+ public void testListPaging() {
+ IPage<AlertGroup> page = new Page<>(1, 10);
page.setTotal(1L);
page.setRecords(getList());
- Mockito.when(alertGroupMapper.queryAlertGroupPage(any(Page.class),eq(groupName))).thenReturn(page);
+ Mockito.when(alertGroupMapper.queryAlertGroupPage(any(Page.class), eq(groupName))).thenReturn(page);
User user = new User();
// no operate
- Map<String, Object> result = alertGroupService.listPaging(user,groupName,1,10);
+ Map<String, Object> result = alertGroupService.listPaging(user, groupName, 1, 10);
logger.info(result.toString());
- Assert.assertEquals(Status.USER_NO_OPERATION_PERM,result.get(Constants.STATUS));
+ Assert.assertEquals(Status.USER_NO_OPERATION_PERM, result.get(Constants.STATUS));
//success
user.setUserType(UserType.ADMIN_USER);
- result = alertGroupService.listPaging(user,groupName,1,10);
+ result = alertGroupService.listPaging(user, groupName, 1, 10);
logger.info(result.toString());
PageInfo<AlertGroup> pageInfo = (PageInfo<AlertGroup>) result.get(Constants.DATA_LIST);
Assert.assertTrue(CollectionUtils.isNotEmpty(pageInfo.getLists()));
}
- @Test
- public void testCreateAlertgroup(){
+ @Test
+ public void testCreateAlertgroup() {
Mockito.when(alertGroupMapper.insert(any(AlertGroup.class))).thenReturn(2);
User user = new User();
//no operate
- Map<String, Object> result = alertGroupService.createAlertgroup(user,groupName, AlertType.EMAIL,groupName);
+ Map<String, Object> result = alertGroupService.createAlertgroup(user, groupName, AlertType.EMAIL, groupName);
logger.info(result.toString());
- Assert.assertEquals(Status.USER_NO_OPERATION_PERM,result.get(Constants.STATUS));
+ Assert.assertEquals(Status.USER_NO_OPERATION_PERM, result.get(Constants.STATUS));
user.setUserType(UserType.ADMIN_USER);
//success
- result = alertGroupService.createAlertgroup(user,groupName, AlertType.EMAIL,groupName);
+ result = alertGroupService.createAlertgroup(user, groupName, AlertType.EMAIL, groupName);
logger.info(result.toString());
- Assert.assertEquals(Status.SUCCESS,result.get(Constants.STATUS));
+ Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
}
+
@Test
- public void testUpdateAlertgroup(){
+ public void testUpdateAlertgroup() {
User user = new User();
// no operate
- Map<String, Object> result = alertGroupService.updateAlertgroup(user,1,groupName, AlertType.SMS,groupName);
+ Map<String, Object> result = alertGroupService.updateAlertgroup(user, 1, groupName, AlertType.SMS, groupName);
logger.info(result.toString());
- Assert.assertEquals(Status.USER_NO_OPERATION_PERM,result.get(Constants.STATUS));
+ Assert.assertEquals(Status.USER_NO_OPERATION_PERM, result.get(Constants.STATUS));
user.setUserType(UserType.ADMIN_USER);
// not exist
- result = alertGroupService.updateAlertgroup(user,1,groupName, AlertType.SMS,groupName);
+ result = alertGroupService.updateAlertgroup(user, 1, groupName, AlertType.SMS, groupName);
logger.info(result.toString());
- Assert.assertEquals(Status.ALERT_GROUP_NOT_EXIST,result.get(Constants.STATUS));
+ Assert.assertEquals(Status.ALERT_GROUP_NOT_EXIST, result.get(Constants.STATUS));
//success
Mockito.when(alertGroupMapper.selectById(2)).thenReturn(getEntity());
- result = alertGroupService.updateAlertgroup(user,2,groupName, AlertType.SMS,groupName);
+ result = alertGroupService.updateAlertgroup(user, 2, groupName, AlertType.SMS, groupName);
logger.info(result.toString());
- Assert.assertEquals(Status.SUCCESS,result.get(Constants.STATUS));
+ Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
}
+
@Test
- public void testDelAlertgroupById(){
+ public void testDelAlertgroupById() {
User user = new User();
// no operate
- Map<String, Object> result = alertGroupService.delAlertgroupById(user,1);
+ Map<String, Object> result = alertGroupService.delAlertgroupById(user, 1);
logger.info(result.toString());
- Assert.assertEquals(Status.USER_NO_OPERATION_PERM,result.get(Constants.STATUS));
+ Assert.assertEquals(Status.USER_NO_OPERATION_PERM, result.get(Constants.STATUS));
user.setUserType(UserType.ADMIN_USER);
// not exist
- result = alertGroupService.delAlertgroupById(user,2);
+ result = alertGroupService.delAlertgroupById(user, 2);
logger.info(result.toString());
- Assert.assertEquals(Status.ALERT_GROUP_NOT_EXIST,result.get(Constants.STATUS));
+ Assert.assertEquals(Status.ALERT_GROUP_NOT_EXIST, result.get(Constants.STATUS));
//success
Mockito.when(alertGroupMapper.selectById(2)).thenReturn(getEntity());
- result = alertGroupService.delAlertgroupById(user,2);
+ result = alertGroupService.delAlertgroupById(user, 2);
logger.info(result.toString());
- Assert.assertEquals(Status.SUCCESS,result.get(Constants.STATUS));
-
+ Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
}
@Test
- public void testGrantUser() {
-
- Integer groupId = 1;
-
- ArgumentCaptor<Integer> groupArgument = ArgumentCaptor.forClass(Integer.class);
-
- Mockito.when(userAlertGroupService.deleteByAlertGroupId(anyInt())).thenReturn(true);
-
- Map<String, Object> result = alertGroupService.grantUser(getLoginUser(), groupId, "123,321");
- Mockito.verify(userAlertGroupService).deleteByAlertGroupId(groupArgument.capture());
-
- logger.info(result.toString());
- assertEquals(groupArgument.getValue(), groupId);
- assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
- }
-
- @Test
public void testVerifyGroupName() {
//group name not exist
boolean result = alertGroupService.existGroupName(groupName);
@@ -190,12 +160,12 @@ public class AlertGroupServiceTest {
Assert.assertTrue(result);
}
-
/**
* create admin user
+ *
* @return
*/
- private User getLoginUser(){
+ private User getLoginUser() {
User loginUser = new User();
loginUser.setUserType(UserType.ADMIN_USER);
@@ -205,9 +175,10 @@ public class AlertGroupServiceTest {
/**
* get list
+ *
* @return
*/
- private List<AlertGroup> getList(){
+ private List<AlertGroup> getList() {
List<AlertGroup> alertGroups = new ArrayList<>();
alertGroups.add(getEntity());
return alertGroups;
@@ -215,13 +186,13 @@ public class AlertGroupServiceTest {
/**
* get entity
+ *
* @return
*/
- private AlertGroup getEntity(){
+ private AlertGroup getEntity() {
AlertGroup alertGroup = new AlertGroup();
alertGroup.setId(1);
alertGroup.setGroupName(groupName);
- alertGroup.setGroupType(AlertType.EMAIL);
return alertGroup;
}
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorService2Test.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorService2Test.java
index a4c0c6b..4c87434 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorService2Test.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorService2Test.java
@@ -14,6 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.api.service;
import static org.mockito.ArgumentMatchers.any;
@@ -126,18 +127,17 @@ public class ExecutorService2Test {
*/
@Test
public void testNoComplement() throws ParseException {
- try {
- Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
- Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
- processDefinitionId, cronTime, CommandType.START_PROCESS,
- null, null,
- null, null, 0,
- "", "", RunMode.RUN_MODE_SERIAL,
- Priority.LOW, Constants.DEFAULT_WORKER_GROUP, 110);
- Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
- verify(processService, times(1)).createCommand(any(Command.class));
- } catch (Exception e) {
- }
+
+ Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
+ Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
+ processDefinitionId, cronTime, CommandType.START_PROCESS,
+ null, null,
+ null, null, 0,
+ RunMode.RUN_MODE_SERIAL,
+ Priority.LOW, Constants.DEFAULT_WORKER_GROUP, 110);
+ Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
+ verify(processService, times(1)).createCommand(any(Command.class));
+
}
/**
@@ -145,18 +145,16 @@ public class ExecutorService2Test {
*/
@Test
public void testDateError() throws ParseException {
- try {
- Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
- Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
- processDefinitionId, "2020-01-31 23:00:00,2020-01-01 00:00:00", CommandType.COMPLEMENT_DATA,
- null, null,
- null, null, 0,
- "", "", RunMode.RUN_MODE_SERIAL,
- Priority.LOW, Constants.DEFAULT_WORKER_GROUP, 110);
- Assert.assertEquals(Status.START_PROCESS_INSTANCE_ERROR, result.get(Constants.STATUS));
- verify(processService, times(0)).createCommand(any(Command.class));
- } catch (Exception e) {
- }
+
+ Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
+ Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
+ processDefinitionId, "2020-01-31 23:00:00,2020-01-01 00:00:00", CommandType.COMPLEMENT_DATA,
+ null, null,
+ null, null, 0,
+ RunMode.RUN_MODE_SERIAL,
+ Priority.LOW, Constants.DEFAULT_WORKER_GROUP, 110);
+ Assert.assertEquals(Status.START_PROCESS_INSTANCE_ERROR, result.get(Constants.STATUS));
+ verify(processService, times(0)).createCommand(any(Command.class));
}
/**
@@ -164,18 +162,16 @@ public class ExecutorService2Test {
*/
@Test
public void testSerial() throws ParseException {
- try {
- Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
- Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
- processDefinitionId, cronTime, CommandType.COMPLEMENT_DATA,
- null, null,
- null, null, 0,
- "", "", RunMode.RUN_MODE_SERIAL,
- Priority.LOW, Constants.DEFAULT_WORKER_GROUP, 110);
- Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
- verify(processService, times(1)).createCommand(any(Command.class));
- } catch (Exception e) {
- }
+
+ Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
+ Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
+ processDefinitionId, cronTime, CommandType.COMPLEMENT_DATA,
+ null, null,
+ null, null, 0,
+ RunMode.RUN_MODE_SERIAL,
+ Priority.LOW, Constants.DEFAULT_WORKER_GROUP, 110);
+ Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
+ verify(processService, times(1)).createCommand(any(Command.class));
}
/**
@@ -183,18 +179,17 @@ public class ExecutorService2Test {
*/
@Test
public void testParallelWithOutSchedule() throws ParseException {
- try {
- Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
- Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
- processDefinitionId, cronTime, CommandType.COMPLEMENT_DATA,
- null, null,
- null, null, 0,
- "", "", RunMode.RUN_MODE_PARALLEL,
- Priority.LOW, Constants.DEFAULT_WORKER_GROUP, 110);
- Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
- verify(processService, times(31)).createCommand(any(Command.class));
- } catch (Exception e) {
- }
+
+ Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
+ Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
+ processDefinitionId, cronTime, CommandType.COMPLEMENT_DATA,
+ null, null,
+ null, null, 0,
+ RunMode.RUN_MODE_PARALLEL,
+ Priority.LOW, Constants.DEFAULT_WORKER_GROUP, 110);
+ Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
+ verify(processService, times(31)).createCommand(any(Command.class));
+
}
/**
@@ -202,31 +197,28 @@ public class ExecutorService2Test {
*/
@Test
public void testParallelWithSchedule() throws ParseException {
- try {
- Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(oneSchedulerList());
- Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
- processDefinitionId, cronTime, CommandType.COMPLEMENT_DATA,
- null, null,
- null, null, 0,
- "", "", RunMode.RUN_MODE_PARALLEL,
- Priority.LOW, Constants.DEFAULT_WORKER_GROUP, 110);
- Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
- verify(processService, times(15)).createCommand(any(Command.class));
- } catch (Exception e) {
- }
- }
+ Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(oneSchedulerList());
+ Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
+ processDefinitionId, cronTime, CommandType.COMPLEMENT_DATA,
+ null, null,
+ null, null, 0,
+ RunMode.RUN_MODE_PARALLEL,
+ Priority.LOW, Constants.DEFAULT_WORKER_GROUP, 110);
+ Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
+ verify(processService, times(15)).createCommand(any(Command.class));
+ }
@Test
- public void testNoMsterServers() throws ParseException {
+ public void testNoMasterServers() throws ParseException {
Mockito.when(monitorService.getServerListFromZK(true)).thenReturn(new ArrayList<Server>());
Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
- processDefinitionId, cronTime, CommandType.COMPLEMENT_DATA,
- null, null,
- null, null, 0,
- "", "", RunMode.RUN_MODE_PARALLEL,
- Priority.LOW, Constants.DEFAULT_WORKER_GROUP, 110);
+ processDefinitionId, cronTime, CommandType.COMPLEMENT_DATA,
+ null, null,
+ null, null, 0,
+ RunMode.RUN_MODE_PARALLEL,
+ Priority.LOW, Constants.DEFAULT_WORKER_GROUP, 110);
Assert.assertEquals(result.get(Constants.STATUS), Status.MASTER_NOT_EXISTS);
}
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionVersionServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionVersionServiceTest.java
index 169ef2b..5a8d09f 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionVersionServiceTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionVersionServiceTest.java
@@ -216,18 +216,17 @@ public class ProcessDefinitionVersionServiceTest {
*/
private ProcessDefinitionVersion getProcessDefinitionVersion(ProcessDefinition processDefinition) {
return ProcessDefinitionVersion
- .newBuilder()
- .processDefinitionId(processDefinition.getId())
- .version(1)
- .processDefinitionJson(processDefinition.getProcessDefinitionJson())
- .description(processDefinition.getDescription())
- .locations(processDefinition.getLocations())
- .connects(processDefinition.getConnects())
- .timeout(processDefinition.getTimeout())
- .globalParams(processDefinition.getGlobalParams())
- .createTime(processDefinition.getUpdateTime())
- .receivers(processDefinition.getReceivers())
- .receiversCc(processDefinition.getReceiversCc())
+ .newBuilder()
+ .processDefinitionId(processDefinition.getId())
+ .version(1)
+ .processDefinitionJson(processDefinition.getProcessDefinitionJson())
+ .description(processDefinition.getDescription())
+ .locations(processDefinition.getLocations())
+ .connects(processDefinition.getConnects())
+ .timeout(processDefinition.getTimeout())
+ .globalParams(processDefinition.getGlobalParams())
+ .createTime(processDefinition.getUpdateTime())
+ .warningGroupId(processDefinition.getWarningGroupId())
.resourceIds(processDefinition.getResourceIds())
.build();
}
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/UserAlertGroupServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/UserAlertGroupServiceTest.java
deleted file mode 100644
index 24b1d5a..0000000
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/UserAlertGroupServiceTest.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.dolphinscheduler.api.service;
-
-import org.apache.dolphinscheduler.dao.mapper.UserAlertGroupMapper;
-import static org.junit.Assert.assertEquals;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.ArgumentCaptor;
-import org.mockito.InjectMocks;
-import org.mockito.Mock;
-import org.mockito.Mockito;
-import org.mockito.junit.MockitoJUnitRunner;
-
-/**
- *
- */
-@RunWith(MockitoJUnitRunner.class)
-public class UserAlertGroupServiceTest {
-
- @InjectMocks
- UserAlertGroupService userAlertGroupService;
-
- @Mock
- UserAlertGroupMapper userAlertGroupMapper;
-
- @Test
- public void deleteByAlertGroupId() {
-
- Integer groupId = 1;
- userAlertGroupService.deleteByAlertGroupId(groupId);
- ArgumentCaptor<Integer> argumentCaptor = ArgumentCaptor.forClass(Integer.class);
-
- Mockito.verify(userAlertGroupMapper).deleteByAlertgroupId(argumentCaptor.capture());
- assertEquals(argumentCaptor.getValue(), groupId);
-
- }
-
-}
\ No newline at end of file
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java
index 4a5c7be..d03b7c5 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java
@@ -19,10 +19,8 @@ package org.apache.dolphinscheduler.dao;
import org.apache.dolphinscheduler.common.enums.AlertEvent;
import org.apache.dolphinscheduler.common.enums.AlertStatus;
-import org.apache.dolphinscheduler.common.enums.AlertType;
import org.apache.dolphinscheduler.common.enums.AlertWarnLevel;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
-import org.apache.dolphinscheduler.common.utils.StringUtils;
import org.apache.dolphinscheduler.dao.datasource.ConnectionFactory;
import org.apache.dolphinscheduler.dao.entity.Alert;
import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance;
@@ -30,11 +28,9 @@ import org.apache.dolphinscheduler.dao.entity.ProcessAlertContent;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.ServerAlertContent;
-import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.AlertGroupMapper;
import org.apache.dolphinscheduler.dao.mapper.AlertMapper;
import org.apache.dolphinscheduler.dao.mapper.AlertPluginInstanceMapper;
-import org.apache.dolphinscheduler.dao.mapper.UserAlertGroupMapper;
import java.util.ArrayList;
import java.util.Date;
@@ -54,9 +50,6 @@ public class AlertDao extends AbstractBaseDao {
private AlertMapper alertMapper;
@Autowired
- private UserAlertGroupMapper userAlertGroupMapper;
-
- @Autowired
private AlertPluginInstanceMapper alertPluginInstanceMapper;
@Autowired
@@ -65,7 +58,6 @@ public class AlertDao extends AbstractBaseDao {
@Override
protected void init() {
alertMapper = ConnectionFactory.getInstance().getMapper(AlertMapper.class);
- userAlertGroupMapper = ConnectionFactory.getInstance().getMapper(UserAlertGroupMapper.class);
alertPluginInstanceMapper = ConnectionFactory.getInstance().getMapper(AlertPluginInstanceMapper.class);
alertGroupMapper = ConnectionFactory.getInstance().getMapper(AlertGroupMapper.class);
}
@@ -97,33 +89,22 @@ public class AlertDao extends AbstractBaseDao {
}
/**
- * query user list by alert group id
- *
- * @param alerGroupId alerGroupId
- * @return user list
- */
- public List<User> queryUserByAlertGroupId(int alerGroupId) {
-
- return userAlertGroupMapper.listUserByAlertgroupId(alerGroupId);
- }
-
- /**
* MasterServer or WorkerServer stoped
*
- * @param alertgroupId alertgroupId
+ * @param alertGroupId alertGroupId
* @param host host
* @param serverType serverType
*/
- public void sendServerStopedAlert(int alertgroupId, String host, String serverType) {
+ public void sendServerStopedAlert(int alertGroupId, String host, String serverType) {
Alert alert = new Alert();
List<ServerAlertContent> serverAlertContents = new ArrayList<>(1);
ServerAlertContent serverStopAlertContent = ServerAlertContent.newBuilder().
- type(serverType).host(host).event(AlertEvent.SERVER_DOWN).warningLevel(AlertWarnLevel.SERIOUS).
- build();
+ type(serverType).host(host).event(AlertEvent.SERVER_DOWN).warningLevel(AlertWarnLevel.SERIOUS).
+ build();
serverAlertContents.add(serverStopAlertContent);
String content = JSONUtils.toJsonString(serverAlertContents);
alert.setTitle("Fault tolerance warning");
- saveTaskTimeoutAlert(alert, content, alertgroupId, null, null);
+ saveTaskTimeoutAlert(alert, content, alertGroupId);
}
/**
@@ -133,34 +114,24 @@ public class AlertDao extends AbstractBaseDao {
* @param processDefinition processDefinition
*/
public void sendProcessTimeoutAlert(ProcessInstance processInstance, ProcessDefinition processDefinition) {
- int alertgroupId = processInstance.getWarningGroupId();
- String receivers = processDefinition.getReceivers();
- String receiversCc = processDefinition.getReceiversCc();
+ int alertGroupId = processInstance.getWarningGroupId();
Alert alert = new Alert();
List<ProcessAlertContent> processAlertContentList = new ArrayList<>(1);
ProcessAlertContent processAlertContent = ProcessAlertContent.newBuilder()
.processId(processInstance.getId())
- .processName(processInstance.getName())
- .event(AlertEvent.TIME_OUT)
- .warningLevel(AlertWarnLevel.MIDDLE)
- .build();
+ .processName(processInstance.getName())
+ .event(AlertEvent.TIME_OUT)
+ .warningLevel(AlertWarnLevel.MIDDLE)
+ .build();
processAlertContentList.add(processAlertContent);
String content = JSONUtils.toJsonString(processAlertContentList);
alert.setTitle("Process Timeout Warn");
- saveTaskTimeoutAlert(alert, content, alertgroupId, receivers, receiversCc);
+ saveTaskTimeoutAlert(alert, content, alertGroupId);
}
- private void saveTaskTimeoutAlert(Alert alert, String content, int alertgroupId, String receivers, String receiversCc) {
- //alert.setShowType(ShowType.TABLE);
+ private void saveTaskTimeoutAlert(Alert alert, String content, int alertGroupId) {
+ alert.setAlertGroupId(alertGroupId);
alert.setContent(content);
- alert.setAlertType(AlertType.EMAIL);
- alert.setAlertGroupId(alertgroupId);
- if (StringUtils.isNotEmpty(receivers)) {
- alert.setReceivers(receivers);
- }
- if (StringUtils.isNotEmpty(receiversCc)) {
- alert.setReceiversCc(receiversCc);
- }
alert.setCreateTime(new Date());
alert.setUpdateTime(new Date());
alertMapper.insert(alert);
@@ -169,30 +140,28 @@ public class AlertDao extends AbstractBaseDao {
/**
* task timeout warn
*
- * @param alertgroupId alertgroupId
- * @param receivers receivers
- * @param receiversCc receiversCc
+ * @param alertGroupId alertGroupId
* @param processInstanceId processInstanceId
* @param processInstanceName processInstanceName
* @param taskId taskId
* @param taskName taskName
*/
- public void sendTaskTimeoutAlert(int alertgroupId, String receivers, String receiversCc, int processInstanceId,
+ public void sendTaskTimeoutAlert(int alertGroupId, int processInstanceId,
String processInstanceName, int taskId, String taskName) {
Alert alert = new Alert();
List<ProcessAlertContent> processAlertContentList = new ArrayList<>(1);
ProcessAlertContent processAlertContent = ProcessAlertContent.newBuilder()
- .processId(processInstanceId)
- .processName(processInstanceName)
- .taskId(taskId)
- .taskName(taskName)
- .event(AlertEvent.TIME_OUT)
- .warningLevel(AlertWarnLevel.MIDDLE)
+ .processId(processInstanceId)
+ .processName(processInstanceName)
+ .taskId(taskId)
+ .taskName(taskName)
+ .event(AlertEvent.TIME_OUT)
+ .warningLevel(AlertWarnLevel.MIDDLE)
.build();
processAlertContentList.add(processAlertContent);
String content = JSONUtils.toJsonString(processAlertContentList);
alert.setTitle("Task Timeout Warn");
- saveTaskTimeoutAlert(alert, content, alertgroupId, receivers, receiversCc);
+ saveTaskTimeoutAlert(alert, content, alertGroupId);
}
/**
@@ -205,16 +174,6 @@ public class AlertDao extends AbstractBaseDao {
}
/**
- * list user information by alert group id
- *
- * @param alertgroupId alertgroupId
- * @return user list
- */
- public List<User> listUserByAlertgroupId(int alertgroupId) {
- return userAlertGroupMapper.listUserByAlertgroupId(alertgroupId);
- }
-
- /**
* for test
*
* @return AlertMapper
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java
index 5f0796c..752ab9d 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java
@@ -18,7 +18,6 @@
package org.apache.dolphinscheduler.dao.entity;
import org.apache.dolphinscheduler.common.enums.AlertStatus;
-import org.apache.dolphinscheduler.common.enums.AlertType;
import java.util.Date;
import java.util.HashMap;
@@ -41,23 +40,13 @@ public class Alert {
*/
@TableField(value = "title")
private String title;
- /**
- * show_type
- */
- //TODO ShowType should be delete from Alert, Because showType is move to the plugin params
- @TableField(value = "show_type")
- private String showType;
+
/**
* content
*/
@TableField(value = "content")
private String content;
- /**
- * alert_type
- */
- //TODO alertType should be delete from Alert, because alert type is decide by the AlertPlugin instance
- @TableField(value = "alert_type")
- private AlertType alertType;
+
/**
* alert_status
*/
@@ -73,18 +62,7 @@ public class Alert {
*/
@TableField("alertgroup_id")
private int alertGroupId;
- /**
- * receivers
- */
- //TODO receivers should be delete from Alert, because only email alert need receivers . And receivers is move to Email Alert Plugin params.
- @TableField("receivers")
- private String receivers;
- /**
- * receivers_cc
- */
- //TODO receivers_cc should be delete from Alert, because only email alert need receivers_cc . And receivers_cc is move to Email Alert Plugin params.
- @TableField("receivers_cc")
- private String receiversCc;
+
/**
* create_time
*/
@@ -117,14 +95,6 @@ public class Alert {
this.title = title;
}
- // public String getShowType() {
- // return showType;
- // }
- //
- // public void setShowType(String showType) {
- // this.showType = showType;
- // }
-
public String getContent() {
return content;
}
@@ -133,14 +103,6 @@ public class Alert {
this.content = content;
}
- public AlertType getAlertType() {
- return alertType;
- }
-
- public void setAlertType(AlertType alertType) {
- this.alertType = alertType;
- }
-
public AlertStatus getAlertStatus() {
return alertStatus;
}
@@ -165,22 +127,6 @@ public class Alert {
this.alertGroupId = alertGroupId;
}
- public String getReceivers() {
- return receivers;
- }
-
- public void setReceivers(String receivers) {
- this.receivers = receivers;
- }
-
- public String getReceiversCc() {
- return receiversCc;
- }
-
- public void setReceiversCc(String receiversCc) {
- this.receiversCc = receiversCc;
- }
-
public Date getCreateTime() {
return createTime;
}
@@ -225,27 +171,15 @@ public class Alert {
if (!title.equals(alert.title)) {
return false;
}
- if (showType != alert.showType) {
- return false;
- }
if (!content.equals(alert.content)) {
return false;
}
- if (alertType != alert.alertType) {
- return false;
- }
if (alertStatus != alert.alertStatus) {
return false;
}
if (!log.equals(alert.log)) {
return false;
}
- if (!receivers.equals(alert.receivers)) {
- return false;
- }
- if (!receiversCc.equals(alert.receiversCc)) {
- return false;
- }
if (!createTime.equals(alert.createTime)) {
return false;
}
@@ -257,14 +191,10 @@ public class Alert {
public int hashCode() {
int result = id;
result = 31 * result + title.hashCode();
- result = 31 * result + showType.hashCode();
result = 31 * result + content.hashCode();
- result = 31 * result + alertType.hashCode();
result = 31 * result + alertStatus.hashCode();
result = 31 * result + log.hashCode();
result = 31 * result + alertGroupId;
- result = 31 * result + receivers.hashCode();
- result = 31 * result + receiversCc.hashCode();
result = 31 * result + createTime.hashCode();
result = 31 * result + updateTime.hashCode();
result = 31 * result + info.hashCode();
@@ -278,13 +208,9 @@ public class Alert {
+ id
+ ", title='"
+ title + '\''
- + ", showType="
- + showType
+ ", content='"
+ content
+ '\''
- + ", alertType="
- + alertType
+ ", alertStatus="
+ alertStatus
+ ", log='"
@@ -292,11 +218,6 @@ public class Alert {
+ '\''
+ ", alertGroupId="
+ alertGroupId
- + ", receivers='"
- + receivers
- + '\''
- + ", receiversCc='"
- + receiversCc
+ '\''
+ ", createTime="
+ createTime
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/AlertGroup.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/AlertGroup.java
index e35e2f7..93eedf6 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/AlertGroup.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/AlertGroup.java
@@ -14,16 +14,16 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.dao.entity;
-import com.baomidou.mybatisplus.annotation.TableField;
-import org.apache.dolphinscheduler.common.enums.AlertType;
+import java.util.Date;
+
import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
-import java.util.Date;
-
@TableName("t_ds_alertgroup")
public class AlertGroup {
/**
@@ -36,11 +36,7 @@ public class AlertGroup {
*/
@TableField(value = "group_name")
private String groupName;
- /**
- * group_type
- */
- @TableField(value = "group_type")
- private AlertType groupType;
+
/**
* description
*/
@@ -57,6 +53,12 @@ public class AlertGroup {
@TableField(value = "update_time")
private Date updateTime;
+ /**
+ * create_user_id
+ */
+ @TableField(value = "create_user_id")
+ private int createUserId;
+
public int getId() {
return id;
}
@@ -73,14 +75,6 @@ public class AlertGroup {
this.groupName = groupName;
}
- public AlertType getGroupType() {
- return groupType;
- }
-
- public void setGroupType(AlertType groupType) {
- this.groupType = groupType;
- }
-
public Date getCreateTime() {
return createTime;
}
@@ -105,6 +99,14 @@ public class AlertGroup {
this.description = description;
}
+ public int getCreateUserId() {
+ return createUserId;
+ }
+
+ public void setCreateUserId(int createUserId) {
+ this.createUserId = createUserId;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -119,10 +121,10 @@ public class AlertGroup {
if (id != that.id) {
return false;
}
- if (groupName != null ? !groupName.equals(that.groupName) : that.groupName != null) {
+ if (createUserId != that.createUserId) {
return false;
}
- if (groupType != that.groupType) {
+ if (groupName != null ? !groupName.equals(that.groupName) : that.groupName != null) {
return false;
}
if (description != null ? !description.equals(that.description) : that.description != null) {
@@ -135,8 +137,8 @@ public class AlertGroup {
@Override
public int hashCode() {
int result = id;
+ result = 31 * result + createUserId;
result = 31 * result + (groupName != null ? groupName.hashCode() : 0);
- result = 31 * result + (groupType != null ? groupType.hashCode() : 0);
result = 31 * result + (description != null ? description.hashCode() : 0);
result = 31 * result + (createTime != null ? createTime.hashCode() : 0);
result = 31 * result + (updateTime != null ? updateTime.hashCode() : 0);
@@ -145,13 +147,13 @@ public class AlertGroup {
@Override
public String toString() {
- return "AlertGroup{" +
- "id=" + id +
- ", groupName='" + groupName + '\'' +
- ", groupType=" + groupType +
- ", description='" + description + '\'' +
- ", createTime=" + createTime +
- ", updateTime=" + updateTime +
- '}';
+ return "AlertGroup{"
+ + "id=" + id
+ + "createUserId=" + createUserId
+ + ", groupName='" + groupName + '\''
+ + ", description='" + description + '\''
+ + ", createTime=" + createTime
+ + ", updateTime=" + updateTime
+ + '}';
}
}
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/ProcessDefinition.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/ProcessDefinition.java
index 56f6cfe..0c2b844 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/ProcessDefinition.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/ProcessDefinition.java
@@ -140,16 +140,6 @@ public class ProcessDefinition {
private String connects;
/**
- * receivers
- */
- private String receivers;
-
- /**
- * receivers cc
- */
- private String receiversCc;
-
- /**
* schedule release state : online/offline
*/
@TableField(exist = false)
@@ -175,6 +165,11 @@ public class ProcessDefinition {
*/
private String resourceIds;
+ /**
+ * warningGroupId
+ */
+ @TableField(exist = false)
+ private int warningGroupId;
public String getName() {
return name;
@@ -272,7 +267,6 @@ public class ProcessDefinition {
this.projectName = projectName;
}
-
public String getGlobalParams() {
return globalParams;
}
@@ -324,22 +318,6 @@ public class ProcessDefinition {
this.connects = connects;
}
- public String getReceivers() {
- return receivers;
- }
-
- public void setReceivers(String receivers) {
- this.receivers = receivers;
- }
-
- public String getReceiversCc() {
- return receiversCc;
- }
-
- public void setReceiversCc(String receiversCc) {
- this.receiversCc = receiversCc;
- }
-
public ReleaseState getScheduleReleaseState() {
return scheduleReleaseState;
}
@@ -388,35 +366,42 @@ public class ProcessDefinition {
this.modifyBy = modifyBy;
}
+ public int getWarningGroupId() {
+ return warningGroupId;
+ }
+
+ public void setWarningGroupId(int warningGroupId) {
+ this.warningGroupId = warningGroupId;
+ }
+
@Override
public String toString() {
- return "ProcessDefinition{" +
- "id=" + id +
- ", name='" + name + '\'' +
- ", version=" + version +
- ", releaseState=" + releaseState +
- ", projectId=" + projectId +
- ", processDefinitionJson='" + processDefinitionJson + '\'' +
- ", description='" + description + '\'' +
- ", globalParams='" + globalParams + '\'' +
- ", globalParamList=" + globalParamList +
- ", globalParamMap=" + globalParamMap +
- ", createTime=" + createTime +
- ", updateTime=" + updateTime +
- ", flag=" + flag +
- ", userId=" + userId +
- ", userName='" + userName + '\'' +
- ", projectName='" + projectName + '\'' +
- ", locations='" + locations + '\'' +
- ", connects='" + connects + '\'' +
- ", receivers='" + receivers + '\'' +
- ", receiversCc='" + receiversCc + '\'' +
- ", scheduleReleaseState=" + scheduleReleaseState +
- ", timeout=" + timeout +
- ", tenantId=" + tenantId +
- ", modifyBy='" + modifyBy + '\'' +
- ", resourceIds='" + resourceIds + '\'' +
- '}';
+ return "ProcessDefinition{"
+ + "id=" + id
+ + ", name='" + name + '\''
+ + ", version=" + version
+ + ", releaseState=" + releaseState
+ + ", projectId=" + projectId
+ + ", processDefinitionJson='" + processDefinitionJson + '\''
+ + ", description='" + description + '\''
+ + ", globalParams='" + globalParams + '\''
+ + ", globalParamList=" + globalParamList
+ + ", globalParamMap=" + globalParamMap
+ + ", createTime=" + createTime
+ + ", updateTime=" + updateTime
+ + ", flag=" + flag
+ + ", userId=" + userId
+ + ", userName='" + userName + '\''
+ + ", projectName='" + projectName + '\''
+ + ", locations='" + locations + '\''
+ + ", connects='" + connects + '\''
+ + ", scheduleReleaseState=" + scheduleReleaseState
+ + ", timeout=" + timeout
+ + ", warningGroupId=" + warningGroupId
+ + ", tenantId=" + tenantId
+ + ", modifyBy='" + modifyBy + '\''
+ + ", resourceIds='" + resourceIds + '\''
+ + '}';
}
}
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/ProcessDefinitionVersion.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/ProcessDefinitionVersion.java
index 26779ba..1c4d979 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/ProcessDefinitionVersion.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/ProcessDefinitionVersion.java
@@ -20,6 +20,7 @@ package org.apache.dolphinscheduler.dao.entity;
import java.util.Date;
import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import com.fasterxml.jackson.annotation.JsonFormat;
@@ -58,16 +59,6 @@ public class ProcessDefinitionVersion {
private String description;
/**
- * receivers
- */
- private String receivers;
-
- /**
- * receivers cc
- */
- private String receiversCc;
-
- /**
* process warning time out. unit: minute
*/
private int timeout;
@@ -98,6 +89,13 @@ public class ProcessDefinitionVersion {
*/
private String connects;
+
+ /**
+ * warningGroupId
+ */
+ @TableField(exist = false)
+ private int warningGroupId;
+
public String getGlobalParams() {
return globalParams;
}
@@ -170,22 +168,6 @@ public class ProcessDefinitionVersion {
this.connects = connects;
}
- public String getReceivers() {
- return receivers;
- }
-
- public void setReceivers(String receivers) {
- this.receivers = receivers;
- }
-
- public String getReceiversCc() {
- return receiversCc;
- }
-
- public void setReceiversCc(String receiversCc) {
- this.receiversCc = receiversCc;
- }
-
public int getTimeout() {
return timeout;
}
@@ -202,23 +184,30 @@ public class ProcessDefinitionVersion {
this.resourceIds = resourceIds;
}
+ public int getWarningGroupId() {
+ return warningGroupId;
+ }
+
+ public void setWarningGroupId(int warningGroupId) {
+ this.warningGroupId = warningGroupId;
+ }
+
@Override
public String toString() {
return "ProcessDefinitionVersion{"
- + "id=" + id
- + ", processDefinitionId=" + processDefinitionId
- + ", version=" + version
- + ", processDefinitionJson='" + processDefinitionJson + '\''
- + ", description='" + description + '\''
- + ", globalParams='" + globalParams + '\''
- + ", createTime=" + createTime
- + ", locations='" + locations + '\''
- + ", connects='" + connects + '\''
- + ", receivers='" + receivers + '\''
- + ", receiversCc='" + receiversCc + '\''
- + ", timeout=" + timeout
- + ", resourceIds='" + resourceIds + '\''
- + '}';
+ + "id=" + id
+ + ", processDefinitionId=" + processDefinitionId
+ + ", version=" + version
+ + ", processDefinitionJson='" + processDefinitionJson + '\''
+ + ", description='" + description + '\''
+ + ", globalParams='" + globalParams + '\''
+ + ", createTime=" + createTime
+ + ", locations='" + locations + '\''
+ + ", connects='" + connects + '\''
+ + ", timeout=" + timeout
+ + ", warningGroupId=" + warningGroupId
+ + ", resourceIds='" + resourceIds + '\''
+ + '}';
}
public static Builder newBuilder() {
@@ -235,9 +224,8 @@ public class ProcessDefinitionVersion {
private Date createTime;
private String locations;
private String connects;
- private String receivers;
- private String receiversCc;
private int timeout;
+ private int warningGroupId;
private String resourceIds;
private Builder() {
@@ -288,18 +276,13 @@ public class ProcessDefinitionVersion {
return this;
}
- public Builder receivers(String receivers) {
- this.receivers = receivers;
- return this;
- }
-
- public Builder receiversCc(String receiversCc) {
- this.receiversCc = receiversCc;
+ public Builder timeout(int timeout) {
+ this.timeout = timeout;
return this;
}
- public Builder timeout(int timeout) {
- this.timeout = timeout;
+ public Builder warningGroupId(int warningGroupId) {
+ this.warningGroupId = warningGroupId;
return this;
}
@@ -319,9 +302,8 @@ public class ProcessDefinitionVersion {
processDefinitionVersion.setCreateTime(createTime);
processDefinitionVersion.setLocations(locations);
processDefinitionVersion.setConnects(connects);
- processDefinitionVersion.setReceivers(receivers);
- processDefinitionVersion.setReceiversCc(receiversCc);
processDefinitionVersion.setTimeout(timeout);
+ processDefinitionVersion.setWarningGroupId(warningGroupId);
processDefinitionVersion.setResourceIds(resourceIds);
return processDefinitionVersion;
}
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/UserAlertGroup.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/UserAlertGroup.java
deleted file mode 100644
index 902196d..0000000
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/UserAlertGroup.java
+++ /dev/null
@@ -1,142 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.dolphinscheduler.dao.entity;
-
-import com.baomidou.mybatisplus.annotation.IdType;
-import com.baomidou.mybatisplus.annotation.TableField;
-import com.baomidou.mybatisplus.annotation.TableId;
-import com.baomidou.mybatisplus.annotation.TableName;
-import com.fasterxml.jackson.annotation.JsonFormat;
-
-import java.util.Date;
-
-/**
- * user alert group
- */
-@TableName("t_ds_relation_user_alertgroup")
-public class UserAlertGroup {
-
- /**
- * id
- */
- @TableId(value="id", type=IdType.AUTO)
- private int id;
-
- /**
- * id
- */
- @TableField("alertgroup_id")
- private int alertgroupId;
-
- /**
- * alert group name
- */
- @TableField(exist = false)
- private String alertgroupName;
-
- /**
- * user id
- */
- private int userId;
-
- /**
- * user name
- */
- @TableField(exist = false)
- private String userName;
-
- /**
- * create time
- */
- @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss",timezone="GMT+8")
- private Date createTime;
-
- /**
- * update time
- */
- @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss",timezone="GMT+8")
- private Date updateTime;
-
- public int getId() {
- return id;
- }
-
- public void setId(int id) {
- this.id = id;
- }
-
- public int getAlertgroupId() {
- return alertgroupId;
- }
-
- public void setAlertgroupId(int alertgroupId) {
- this.alertgroupId = alertgroupId;
- }
-
- public int getUserId() {
- return userId;
- }
-
- public void setUserId(int userId) {
- this.userId = userId;
- }
-
- public Date getCreateTime() {
- return createTime;
- }
-
- public void setCreateTime(Date createTime) {
- this.createTime = createTime;
- }
-
- public Date getUpdateTime() {
- return updateTime;
- }
-
- public void setUpdateTime(Date updateTime) {
- this.updateTime = updateTime;
- }
-
- public String getAlertgroupName() {
- return alertgroupName;
- }
-
- public void setAlertgroupName(String alertgroupName) {
- this.alertgroupName = alertgroupName;
- }
-
- public String getUserName() {
- return userName;
- }
-
- public void setUserName(String userName) {
- this.userName = userName;
- }
-
- @Override
- public String toString() {
- return "UserAlertGroup{" +
- "id=" + id +
- ", alertgroupId=" + alertgroupId +
- ", alertgroupName='" + alertgroupName + '\'' +
- ", userId=" + userId +
- ", userName='" + userName + '\'' +
- ", createTime=" + createTime +
- ", updateTime=" + updateTime +
- '}';
- }
-}
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertGroupMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertGroupMapper.java
index 0fd26b9..e337ab6 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertGroupMapper.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertGroupMapper.java
@@ -55,14 +55,6 @@ public interface AlertGroupMapper extends BaseMapper<AlertGroup> {
*/
List<AlertGroup> queryByUserId(@Param("userId") int userId);
-
- /**
- * query by alert type
- * @param alertType alertType
- * @return alertgroup list
- */
- List<AlertGroup> queryByAlertType(@Param("alertType") AlertType alertType);
-
/**
* query all group list
* @return alertgroup list
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/UserAlertGroupMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/UserAlertGroupMapper.java
deleted file mode 100644
index f374c20..0000000
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/UserAlertGroupMapper.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.dolphinscheduler.dao.mapper;
-
-import org.apache.dolphinscheduler.dao.entity.User;
-import org.apache.dolphinscheduler.dao.entity.UserAlertGroup;
-import com.baomidou.mybatisplus.core.mapper.BaseMapper;
-import org.apache.ibatis.annotations.Param;
-
-import java.util.List;
-
-/**
- * user alertgroup relation mapper interface
- */
-public interface UserAlertGroupMapper extends BaseMapper<UserAlertGroup> {
-
- /**
- * delete user alertgroup relation by alertgroupId
- * @param alertgroupId alertgroupId
- * @return delete result
- */
- int deleteByAlertgroupId(@Param("alertgroupId") int alertgroupId);
-
- /**
- * list user by alertgroupId
- * @param alertgroupId alertgroupId
- * @return user list
- */
- List<User> listUserByAlertgroupId(@Param("alertgroupId") int alertgroupId);
-
-}
diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertGroupMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertGroupMapper.xml
index 8ee335b..d68d459 100644
--- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertGroupMapper.xml
+++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertGroupMapper.xml
@@ -31,14 +31,10 @@
where group_name=#{groupName}
</select>
<select id="queryByUserId" resultType="org.apache.dolphinscheduler.dao.entity.AlertGroup">
- select * from t_ds_alertgroup t
- left join t_ds_relation_user_alertgroup r on t.id=r.alertgroup_id
- where r.user_id=#{userId}
- </select>
- <select id="queryByAlertType" resultType="org.apache.dolphinscheduler.dao.entity.AlertGroup">
select * from t_ds_alertgroup
- where group_type=#{alertType}
+ where create_user_id = #{userId}
</select>
+
<select id="queryAllGroupList" resultType="org.apache.dolphinscheduler.dao.entity.AlertGroup">
select *
from t_ds_alertgroup
diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionMapper.xml
index 0481f7d..1dc9bbf 100644
--- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionMapper.xml
+++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionMapper.xml
@@ -29,12 +29,14 @@
and pd.name = #{processDefinitionName}
</select>
<select id="queryDefineListPaging" resultType="org.apache.dolphinscheduler.dao.entity.ProcessDefinition">
- SELECT td.id, td.name, td.version, td.release_state, td.project_id, td.user_id, td.description, td.global_params,
- td.flag, td.receivers, td.receivers_cc, td.timeout, td.tenant_id, td.modify_by, td.update_time, td.create_time,
- sc.schedule_release_state, tu.user_name
+ SELECT td.id, td.name, td.version, td.release_state, td.project_id, td.user_id, td.description,
+ td.global_params,
+ td.flag, td.warning_group_id, td.timeout, td.tenant_id, td.modify_by, td.update_time, td.create_time,
+ sc.schedule_release_state, tu.user_name
FROM t_ds_process_definition td
- left join (select process_definition_id,release_state as schedule_release_state from t_ds_schedules group by process_definition_id,release_state) sc on sc.process_definition_id = td.id
- left join t_ds_user tu on td.user_id = tu.id
+ left join (select process_definition_id,release_state as schedule_release_state from t_ds_schedules group by
+ process_definition_id,release_state) sc on sc.process_definition_id = td.id
+ left join t_ds_user tu on td.user_id = tu.id
where td.project_id = #{projectId}
<if test=" searchVal != null and searchVal != ''">
and td.name like concat('%', #{searchVal}, '%')
diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/UserAlertGroupMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/UserAlertGroupMapper.xml
deleted file mode 100644
index cbb4482..0000000
--- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/UserAlertGroupMapper.xml
+++ /dev/null
@@ -1,31 +0,0 @@
-<?xml version="1.0" encoding="UTF-8" ?>
-<!--
- ~ Licensed to the Apache Software Foundation (ASF) under one or more
- ~ contributor license agreements. See the NOTICE file distributed with
- ~ this work for additional information regarding copyright ownership.
- ~ The ASF licenses this file to You under the Apache License, Version 2.0
- ~ (the "License"); you may not use this file except in compliance with
- ~ the License. You may obtain a copy of the License at
- ~
- ~ http://www.apache.org/licenses/LICENSE-2.0
- ~
- ~ Unless required by applicable law or agreed to in writing, software
- ~ distributed under the License is distributed on an "AS IS" BASIS,
- ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- ~ See the License for the specific language governing permissions and
- ~ limitations under the License.
- -->
-
-<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
-<mapper namespace="org.apache.dolphinscheduler.dao.mapper.UserAlertGroupMapper">
- <delete id="deleteByAlertgroupId">
- delete from t_ds_relation_user_alertgroup
- where alertgroup_id = #{alertgroupId}
- </delete>
- <select id="listUserByAlertgroupId" resultType="org.apache.dolphinscheduler.dao.entity.User">
- SELECT u.*
- FROM t_ds_relation_user_alertgroup g_u
- JOIN t_ds_user u on g_u.user_id = u.id
- WHERE g_u.alertgroup_id = #{alertgroupId}
- </select>
-</mapper>
\ No newline at end of file
diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/AlertDaoTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/AlertDaoTest.java
index 0b5c516..0137bd5 100644
--- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/AlertDaoTest.java
+++ b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/AlertDaoTest.java
@@ -18,7 +18,6 @@
package org.apache.dolphinscheduler.dao;
import org.apache.dolphinscheduler.common.enums.AlertStatus;
-import org.apache.dolphinscheduler.common.enums.AlertType;
import org.apache.dolphinscheduler.dao.entity.Alert;
import java.util.List;
@@ -33,10 +32,8 @@ public class AlertDaoTest {
AlertDao alertDao = DaoFactory.getDaoInstance(AlertDao.class);
Alert alert = new Alert();
alert.setTitle("Mysql Exception");
- //alert.setShowType(ShowType.TEXT);
- alert.setContent("[\"alarm time：2018-02-05\", \"service name：MYSQL_ALTER\", \"alarm name：MYSQL_ALTER_DUMP\", " +
- "\"get the alarm exception.！，interface error，exception information：timed out\", \"request address：http://blog.csdn.net/dreamInTheWorld/article/details/78539286\"]");
- alert.setAlertType(AlertType.EMAIL);
+ alert.setContent("[\"alarm time：2018-02-05\", \"service name：MYSQL_ALTER\", \"alarm name：MYSQL_ALTER_DUMP\", "
+ + "\"get the alarm exception.！，interface error，exception information：timed out\", \"request address：http://blog.csdn.net/dreamInTheWorld/article/details/78539286\"]");
alert.setAlertGroupId(1);
alert.setAlertStatus(AlertStatus.WAIT_EXECUTION);
alertDao.addAlert(alert);
diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/AlertGroupMapperTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/AlertGroupMapperTest.java
index c8d0953..db42a2f 100644
--- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/AlertGroupMapperTest.java
+++ b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/AlertGroupMapperTest.java
@@ -14,38 +14,36 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.dao.mapper;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
-import org.apache.dolphinscheduler.common.enums.AlertType;
import org.apache.dolphinscheduler.common.utils.DateUtils;
-import org.apache.dolphinscheduler.dao.entity.AccessToken;
import org.apache.dolphinscheduler.dao.entity.AlertGroup;
-import org.apache.dolphinscheduler.dao.entity.UserAlertGroup;
-import com.baomidou.mybatisplus.core.metadata.IPage;
-import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
-import org.junit.Assert;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.test.annotation.Rollback;
-import org.springframework.test.context.junit4.SpringRunner;
-import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.annotation.Rollback;
+import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.transaction.annotation.Transactional;
-import static org.hamcrest.Matchers.*;
-import static org.junit.Assert.*;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
/**
- * AlertGroup mapper test
+ * AlertGroup mapper test
*/
@RunWith(SpringRunner.class)
@SpringBootTest
@@ -53,22 +51,17 @@ import static org.junit.Assert.*;
@Rollback(true)
public class AlertGroupMapperTest {
-
@Autowired
AlertGroupMapper alertGroupMapper;
- @Autowired
- UserAlertGroupMapper userAlertGroupMapper;
-
-
/**
* test insert
*/
@Test
- public void testInsert(){
+ public void testInsert() {
AlertGroup alertGroup = createAlertGroup();
assertNotNull(alertGroup);
- assertThat(alertGroup.getId(),greaterThan(0));
+ assertThat(alertGroup.getId(), greaterThan(0));
}
@@ -99,7 +92,7 @@ public class AlertGroupMapperTest {
Integer offset = 2;
Integer size = 2;
- Map<Integer, AlertGroup> alertGroupMap = createAlertGroups(count,groupName);
+ Map<Integer, AlertGroup> alertGroupMap = createAlertGroups(count, groupName);
Page page = new Page(offset, size);
@@ -109,25 +102,23 @@ public class AlertGroupMapperTest {
assertEquals(alertGroupList.size(), size.intValue());
- for (AlertGroup alertGroup : alertGroupList){
+ for (AlertGroup alertGroup : alertGroupList) {
AlertGroup resultAlertGroup = alertGroupMap.get(alertGroup.getId());
- if (resultAlertGroup != null){
- assertEquals(alertGroup,resultAlertGroup);
+ if (resultAlertGroup != null) {
+ assertEquals(alertGroup, resultAlertGroup);
}
}
-
}
/**
* test update
*/
@Test
- public void testUpdate(){
+ public void testUpdate() {
AlertGroup alertGroup = createAlertGroup();
alertGroup.setGroupName("modify GroupName");
- alertGroup.setGroupType(AlertType.SMS);
alertGroup.setDescription("modify GroupName");
alertGroup.setUpdateTime(DateUtils.getCurrentDate());
@@ -135,16 +126,15 @@ public class AlertGroupMapperTest {
AlertGroup resultAlertGroup = alertGroupMapper.selectById(alertGroup.getId());
- assertEquals(alertGroup,resultAlertGroup);
+ assertEquals(alertGroup, resultAlertGroup);
}
-
/**
* test delete
*/
@Test
- public void testDelete(){
+ public void testDelete() {
AlertGroup alertGroup = createAlertGroup();
@@ -156,7 +146,6 @@ public class AlertGroupMapperTest {
}
-
/**
* test query by groupname
*/
@@ -169,43 +158,10 @@ public class AlertGroupMapperTest {
List<AlertGroup> alertGroupList = alertGroupMapper.queryByGroupName("testGroup");
-
compareAlertGroups(alertGroupMap, alertGroupList);
}
/**
- * test query by userId
- */
- @Test
- public void testQueryByUserId() {
- Integer count = 4;
- Integer userId = 1;
-
- Map<Integer, AlertGroup> alertGroupMap =
- createAlertGroups(count, userId);
-
- List<AlertGroup> alertGroupList =
- alertGroupMapper.queryByUserId(userId);
-
- compareAlertGroups(alertGroupMap,alertGroupList);
-
- }
-
- /**
- * test query by alert type
- */
- @Test
- public void testQueryByAlertType() {
- Integer count = 4;
-
- Map<Integer, AlertGroup> alertGroupMap = createAlertGroups(count);
- List<AlertGroup> alertGroupList = alertGroupMapper.queryByAlertType(AlertType.EMAIL);
-
- compareAlertGroups(alertGroupMap,alertGroupList);
-
- }
-
- /**
* test query all group list
*/
@Test
@@ -215,37 +171,38 @@ public class AlertGroupMapperTest {
List<AlertGroup> alertGroupList = alertGroupMapper.queryAllGroupList();
- compareAlertGroups(alertGroupMap,alertGroupList);
+ compareAlertGroups(alertGroupMap, alertGroupList);
}
-
/**
* compare AlertGruops
- * @param alertGroupMap alertGroupMap
+ *
+ * @param alertGroupMap alertGroupMap
* @param alertGroupList alertGroupList
*/
- private void compareAlertGroups(Map<Integer,AlertGroup> alertGroupMap,
- List<AlertGroup> alertGroupList){
+ private void compareAlertGroups(Map<Integer, AlertGroup> alertGroupMap,
+ List<AlertGroup> alertGroupList) {
// greaterThanOrEqualTo，beacuse maybe db have already exists
- assertThat(alertGroupList.size(),greaterThanOrEqualTo(alertGroupMap.size()));
+ assertThat(alertGroupList.size(), greaterThanOrEqualTo(alertGroupMap.size()));
- for (AlertGroup alertGroup : alertGroupList){
+ for (AlertGroup alertGroup : alertGroupList) {
AlertGroup resultAlertGroup = alertGroupMap.get(alertGroup.getId());
- if (resultAlertGroup != null){
- assertEquals(alertGroup,resultAlertGroup);
+ if (resultAlertGroup != null) {
+ assertEquals(alertGroup, resultAlertGroup);
}
}
}
+
/**
* insert
+ *
* @return AlertGroup
*/
- private AlertGroup createAlertGroup(String groupName){
+ private AlertGroup createAlertGroup(String groupName) {
AlertGroup alertGroup = new AlertGroup();
alertGroup.setGroupName(groupName);
alertGroup.setDescription("alert group 1");
- alertGroup.setGroupType(AlertType.EMAIL);
alertGroup.setCreateTime(DateUtils.getCurrentDate());
alertGroup.setUpdateTime(DateUtils.getCurrentDate());
@@ -257,13 +214,13 @@ public class AlertGroupMapperTest {
/**
* insert
+ *
* @return AlertGroup
*/
- private AlertGroup createAlertGroup(){
+ private AlertGroup createAlertGroup() {
AlertGroup alertGroup = new AlertGroup();
alertGroup.setGroupName("testGroup");
alertGroup.setDescription("testGroup");
- alertGroup.setGroupType(AlertType.EMAIL);
alertGroup.setCreateTime(DateUtils.getCurrentDate());
alertGroup.setUpdateTime(DateUtils.getCurrentDate());
@@ -275,81 +232,41 @@ public class AlertGroupMapperTest {
/**
* create AlertGroups
- * @param count create AlertGroup count
+ *
+ * @param count create AlertGroup count
* @param groupName groupName
* @return AlertGroup map
*/
- private Map<Integer,AlertGroup> createAlertGroups(
- Integer count,String groupName){
- Map<Integer,AlertGroup> alertGroupMap = new HashMap<>();
+ private Map<Integer, AlertGroup> createAlertGroups(
+ Integer count, String groupName) {
+ Map<Integer, AlertGroup> alertGroupMap = new HashMap<>();
- AlertGroup alertGroup = null;
- for (int i = 0 ; i < count; i++){
+ AlertGroup alertGroup = null;
+ for (int i = 0; i < count; i++) {
alertGroup = createAlertGroup(groupName);
- alertGroupMap.put(alertGroup.getId(),alertGroup);
- }
-
- return alertGroupMap;
- }
-
-
- /**
- * create AlertGroups
- * @param count create AlertGroup count
- * @return AlertGroup map
- */
- private Map<Integer,AlertGroup> createAlertGroups(
- Integer count){
- Map<Integer,AlertGroup> alertGroupMap = new HashMap<>();
-
- AlertGroup alertGroup = null;
- for (int i = 0 ; i < count; i++){
- alertGroup = createAlertGroup();
- alertGroupMap.put(alertGroup.getId(),alertGroup);
+ alertGroupMap.put(alertGroup.getId(), alertGroup);
}
return alertGroupMap;
}
-
/**
* create AlertGroups
+ *
* @param count create AlertGroup count
* @return AlertGroup map
*/
- private Map<Integer,AlertGroup> createAlertGroups(
- Integer count,Integer userId){
- Map<Integer,AlertGroup> alertGroupMap = new HashMap<>();
+ private Map<Integer, AlertGroup> createAlertGroups(
+ Integer count) {
+ Map<Integer, AlertGroup> alertGroupMap = new HashMap<>();
- AlertGroup alertGroup = null;
- for (int i = 0 ; i < count; i++){
+ AlertGroup alertGroup = null;
+ for (int i = 0; i < count; i++) {
alertGroup = createAlertGroup();
-
- createUserAlertGroup(userId,alertGroup.getId());
-
- alertGroupMap.put(alertGroup.getId(),alertGroup);
+ alertGroupMap.put(alertGroup.getId(), alertGroup);
}
return alertGroupMap;
}
- /**
- * create AlertGroup
- * @param userId userId
- * @param alertgroupId alertgroupId
- * @return UserAlertGroup
- */
- private UserAlertGroup createUserAlertGroup(
- Integer userId,Integer alertgroupId){
- UserAlertGroup userAlertGroup = new UserAlertGroup();
- userAlertGroup.setAlertgroupId(alertgroupId);
- userAlertGroup.setUserId(userId);
- userAlertGroup.setCreateTime(DateUtils.getCurrentDate());
- userAlertGroup.setUpdateTime(DateUtils.getCurrentDate());
-
- userAlertGroupMapper.insert(userAlertGroup);
-
- return userAlertGroup;
- }
-
}
\ No newline at end of file
diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/AlertMapperTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/AlertMapperTest.java
index 24b9719..63fcca2 100644
--- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/AlertMapperTest.java
+++ b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/AlertMapperTest.java
@@ -23,7 +23,6 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import org.apache.dolphinscheduler.common.enums.AlertStatus;
-import org.apache.dolphinscheduler.common.enums.AlertType;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.dao.entity.Alert;
@@ -162,14 +161,10 @@ public class AlertMapperTest {
*/
private Alert createAlert(AlertStatus alertStatus) {
Alert alert = new Alert();
- //alert.setShowType(ShowType.TABLE);
alert.setTitle("test alert");
alert.setContent("[{'type':'WORKER','host':'192.168.xx.xx','event':'server down','warning level':'serious'}]");
- alert.setAlertType(AlertType.EMAIL);
alert.setAlertStatus(alertStatus);
alert.setLog("success");
- alert.setReceivers("aa@aa.com");
- alert.setReceiversCc("bb@aa.com");
alert.setCreateTime(DateUtils.getCurrentDate());
alert.setUpdateTime(DateUtils.getCurrentDate());
diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/AlertPluginInstanceMapperTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/AlertPluginInstanceMapperTest.java
index 3fa5fcc..f2f39c6 100644
--- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/AlertPluginInstanceMapperTest.java
+++ b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/AlertPluginInstanceMapperTest.java
@@ -17,7 +17,6 @@
package org.apache.dolphinscheduler.dao.mapper;
-import org.apache.dolphinscheduler.common.enums.AlertType;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.dao.entity.AlertGroup;
import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance;
@@ -106,7 +105,6 @@ public class AlertPluginInstanceMapperTest {
AlertGroup alertGroup = new AlertGroup();
alertGroup.setGroupName(groupName);
alertGroup.setDescription("alert group 1");
- alertGroup.setGroupType(AlertType.EMAIL);
alertGroup.setCreateTime(DateUtils.getCurrentDate());
alertGroup.setUpdateTime(DateUtils.getCurrentDate());
diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionMapperTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionMapperTest.java
index c58c92b..636a9ca 100644
--- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionMapperTest.java
+++ b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionMapperTest.java
@@ -77,8 +77,6 @@ public class ProcessDefinitionMapperTest {
processDefinition.setUserId(101);
processDefinition.setUpdateTime(new Date());
processDefinition.setCreateTime(new Date());
-// processDefinition.setGlobalParams("[{\"prop\":\"selenium_global_parameters_1\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"selenium_global_parameters_value_1\"}]");
-
processDefinitionMapper.insert(processDefinition);
return processDefinition;
}
@@ -170,8 +168,6 @@ public class ProcessDefinitionMapperTest {
processDefinition.setCreateTime(new Date());
processDefinition.setTenantId(tenant.getId());
processDefinition.setUserId(user.getId());
-// processDefinition.setGlobalParams("[{\"prop\":\"selenium_global_parameters_1\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"selenium_global_parameters_value_1\"}]");
-//
processDefinitionMapper.insert(processDefinition);
ProcessDefinition processDefinition1 = processDefinitionMapper.queryByDefineName(project.getId(), "def 1");
diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionVersionMapperTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionVersionMapperTest.java
index e825e33..938e26d 100644
--- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionVersionMapperTest.java
+++ b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionVersionMapperTest.java
@@ -76,8 +76,6 @@ public class ProcessDefinitionVersionMapperTest {
processDefinitionVersion.setCreateTime(new Date());
processDefinitionVersion.setLocations(StringUtils.EMPTY);
processDefinitionVersion.setConnects(StringUtils.EMPTY);
- processDefinitionVersion.setReceivers(StringUtils.EMPTY);
- processDefinitionVersion.setReceiversCc(StringUtils.EMPTY);
processDefinitionVersion.setTimeout(10);
processDefinitionVersion.setResourceIds("1,2");
processDefinitionVersionMapper.insert(processDefinitionVersion);
@@ -101,8 +99,6 @@ public class ProcessDefinitionVersionMapperTest {
processDefinitionVersion.setCreateTime(new Date());
processDefinitionVersion.setLocations(StringUtils.EMPTY);
processDefinitionVersion.setConnects(StringUtils.EMPTY);
- processDefinitionVersion.setReceivers(StringUtils.EMPTY);
- processDefinitionVersion.setReceiversCc(StringUtils.EMPTY);
processDefinitionVersion.setTimeout(10);
processDefinitionVersion.setResourceIds("1,2");
processDefinitionVersionMapper.insert(processDefinitionVersion);
diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/UserAlertGroupMapperTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/UserAlertGroupMapperTest.java
deleted file mode 100644
index 2c5024f..0000000
--- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/UserAlertGroupMapperTest.java
+++ /dev/null
@@ -1,196 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.dolphinscheduler.dao.mapper;
-
-
-import org.apache.dolphinscheduler.common.enums.AlertType;
-import org.apache.dolphinscheduler.common.enums.UserType;
-import org.apache.dolphinscheduler.dao.entity.AlertGroup;
-import org.apache.dolphinscheduler.dao.entity.User;
-import org.apache.dolphinscheduler.dao.entity.UserAlertGroup;
-import org.junit.Assert;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.test.annotation.Rollback;
-import org.springframework.test.context.junit4.SpringRunner;
-import org.springframework.transaction.annotation.Transactional;
-
-import java.util.Date;
-import java.util.List;
-
-@RunWith(SpringRunner.class)
-@SpringBootTest
-@Transactional
-@Rollback(true)
-public class UserAlertGroupMapperTest {
-
- @Autowired
- private UserMapper userMapper;
-
- @Autowired
- AlertGroupMapper alertGroupMapper;
-
- @Autowired
- private UserAlertGroupMapper userAlertGroupMapper;
-
- /**
- * insert one UserAlertGroup
- * @param user user
- * @param alertGroup alertGroup
- * @return UserAlertGroup
- */
- private UserAlertGroup insertOne(User user,AlertGroup alertGroup){
- UserAlertGroup userAlertGroup = new UserAlertGroup();
- userAlertGroup.setAlertgroupName(alertGroup.getGroupName());
- userAlertGroup.setAlertgroupId(alertGroup.getId());
- userAlertGroup.setUserId(user.getId());
- userAlertGroup.setCreateTime(new Date());
- userAlertGroup.setUpdateTime(new Date());
- userAlertGroupMapper.insert(userAlertGroup);
- return userAlertGroup;
- }
-
- /**
- * insert one UserAlertGroup
- * @return UserAlertGroup
- */
- private UserAlertGroup insertOne(){
- UserAlertGroup userAlertGroup = new UserAlertGroup();
- userAlertGroup.setAlertgroupName("dolphin_alert_group");
- userAlertGroup.setAlertgroupId(10);
- userAlertGroup.setUserId(4);
- userAlertGroup.setCreateTime(new Date());
- userAlertGroup.setUpdateTime(new Date());
- userAlertGroupMapper.insert(userAlertGroup);
- return userAlertGroup;
- }
-
- /**
- * insert one user
- * @return User
- */
- private User insertOneUser(){
- User user = new User();
- user.setUserName("user1");
- user.setUserPassword("1");
- user.setEmail("xx@123.com");
- user.setUserType(UserType.GENERAL_USER);
- user.setCreateTime(new Date());
- user.setTenantId(1);
- user.setQueue("dolphin");
- user.setUpdateTime(new Date());
- userMapper.insert(user);
- return user;
- }
-
- /**
- * insert one AlertGroup
- * @return AlertGroup
- */
- private AlertGroup insertOneAlertGroup(){
- //insertOne
- AlertGroup alertGroup = new AlertGroup();
- alertGroup.setGroupName("alert group 1");
- alertGroup.setDescription("alert test1");
- alertGroup.setGroupType(AlertType.EMAIL);
-
- alertGroup.setCreateTime(new Date());
- alertGroup.setUpdateTime(new Date());
- alertGroupMapper.insert(alertGroup);
- return alertGroup;
- }
-
- /**
- * test update
- */
- @Test
- public void testUpdate(){
- //insertOneUser
- User user = insertOneUser();
- //insertOneAlertGroup
- AlertGroup alertGroup = insertOneAlertGroup();
-
- //insertOne
- UserAlertGroup userAlertGroup = insertOne();
- //update
- userAlertGroup.setUserId(user.getId());
- userAlertGroup.setAlertgroupId(alertGroup.getId());
- userAlertGroup.setUpdateTime(new Date());
-
- int update = userAlertGroupMapper.updateById(userAlertGroup);
- Assert.assertEquals(update, 1);
- }
-
- /**
- * test delete
- */
- @Test
- public void testDelete(){
- //insertOne
- UserAlertGroup userAlertGroup = insertOne();
- //delete
- int delete = userAlertGroupMapper.deleteById(userAlertGroup.getId());
- Assert.assertEquals(delete, 1);
- }
-
- /**
- * test query
- */
- @Test
- public void testQuery() {
- //insertOne
- UserAlertGroup userAlertGroup = insertOne();
- //query
- List<UserAlertGroup> userAlertGroupList = userAlertGroupMapper.selectList(null);
- Assert.assertNotEquals(userAlertGroupList.size(), 0);
- }
-
- /**
- * test delete by alertgroupId
- */
- @Test
- public void testDeleteByAlertgroupId() {
- //insertOneUser
- User user = insertOneUser();
- //insertOneAlertGroup
- AlertGroup alertGroup = insertOneAlertGroup();
-
- //insertOne
- UserAlertGroup userAlertGroup = insertOne(user,alertGroup);
- int delete = userAlertGroupMapper.deleteByAlertgroupId(alertGroup.getId());
- Assert.assertEquals(delete, 1);
- }
-
- /**
- * test list user by alertgroupId
- */
- @Test
- public void testListUserByAlertgroupId() {
- //insertOneUser
- User user = insertOneUser();
- //insertOneAlertGroup
- AlertGroup alertGroup = insertOneAlertGroup();
-
- //insertOne
- UserAlertGroup userAlertGroup = insertOne(user,alertGroup);
- List<User> userList = userAlertGroupMapper.listUserByAlertgroupId(alertGroup.getId());
- Assert.assertNotEquals(userList.size(), 0);
-
- }
-}
\ No newline at end of file
diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/UserMapperTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/UserMapperTest.java
index 7b1849e..d037c44 100644
--- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/UserMapperTest.java
+++ b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/UserMapperTest.java
@@ -14,14 +14,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.dao.mapper;
-import com.baomidou.mybatisplus.core.metadata.IPage;
-import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
-import org.apache.dolphinscheduler.common.enums.AlertType;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.utils.DateUtils;
-import org.apache.dolphinscheduler.dao.entity.*;
+import org.apache.dolphinscheduler.dao.entity.AccessToken;
+import org.apache.dolphinscheduler.dao.entity.AlertGroup;
+import org.apache.dolphinscheduler.dao.entity.Queue;
+import org.apache.dolphinscheduler.dao.entity.Tenant;
+import org.apache.dolphinscheduler.dao.entity.User;
+
+import java.util.Date;
+import java.util.List;
+
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -31,8 +37,8 @@ import org.springframework.test.annotation.Rollback;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.transaction.annotation.Transactional;
-import java.util.Date;
-import java.util.List;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
@RunWith(SpringRunner.class)
@SpringBootTest
@@ -46,9 +52,6 @@ public class UserMapperTest {
AlertGroupMapper alertGroupMapper;
@Autowired
- private UserAlertGroupMapper userAlertGroupMapper;
-
- @Autowired
AccessTokenMapper accessTokenMapper;
@Autowired
@@ -59,9 +62,10 @@ public class UserMapperTest {
/**
* insert one user
+ *
* @return User
*/
- private User insertOne(){
+ private User insertOne() {
User user = new User();
user.setUserName("user1");
user.setUserPassword("1");
@@ -76,10 +80,11 @@ public class UserMapperTest {
/**
* insert one user
+ *
* @param tenant tenant
* @return User
*/
- private User insertOne(Tenant tenant){
+ private User insertOne(Tenant tenant) {
User user = new User();
user.setUserName("user1");
user.setUserPassword("1");
@@ -94,11 +99,12 @@ public class UserMapperTest {
/**
* insert one user
- * @param queue queue
+ *
+ * @param queue queue
* @param tenant tenant
* @return User
*/
- private User insertOne(Queue queue,Tenant tenant){
+ private User insertOne(Queue queue, Tenant tenant) {
User user = new User();
user.setUserName("user1");
user.setUserPassword("1");
@@ -114,14 +120,14 @@ public class UserMapperTest {
/**
* insert one AlertGroup
+ *
* @return AlertGroup
*/
- private AlertGroup insertOneAlertGroup(){
+ private AlertGroup insertOneAlertGroup() {
//insertOne
AlertGroup alertGroup = new AlertGroup();
alertGroup.setGroupName("alert group 1");
alertGroup.setDescription("alert test1");
- alertGroup.setGroupType(AlertType.EMAIL);
alertGroup.setCreateTime(new Date());
alertGroup.setUpdateTime(new Date());
@@ -130,44 +136,29 @@ public class UserMapperTest {
}
/**
- * insert one UserAlertGroup
- * @param user user
- * @param alertGroup alertGroup
- * @return UserAlertGroup
- */
- private UserAlertGroup insertOneUserAlertGroup(User user,AlertGroup alertGroup){
- UserAlertGroup userAlertGroup = new UserAlertGroup();
- userAlertGroup.setAlertgroupName(alertGroup.getGroupName());
- userAlertGroup.setAlertgroupId(alertGroup.getId());
- userAlertGroup.setUserId(user.getId());
- userAlertGroup.setCreateTime(new Date());
- userAlertGroup.setUpdateTime(new Date());
- userAlertGroupMapper.insert(userAlertGroup);
- return userAlertGroup;
- }
-
- /**
* insert one AccessToken
+ *
* @param user user
* @return AccessToken
*/
- private AccessToken insertOneAccessToken(User user){
+ private AccessToken insertOneAccessToken(User user) {
//insertOne
AccessToken accessToken = new AccessToken();
accessToken.setUserId(user.getId());
accessToken.setToken("secrettoken");
accessToken.setCreateTime(new Date());
accessToken.setUpdateTime(new Date());
- accessToken.setExpireTime(DateUtils.getSomeHourOfDay(new Date(),1));
+ accessToken.setExpireTime(DateUtils.getSomeHourOfDay(new Date(), 1));
accessTokenMapper.insert(accessToken);
return accessToken;
}
/**
* insert one Tenant
+ *
* @return Tenant
*/
- private Tenant insertOneTenant(){
+ private Tenant insertOneTenant() {
Tenant tenant = new Tenant();
tenant.setTenantCode("dolphin");
tenant.setTenantName("dolphin test");
@@ -181,9 +172,10 @@ public class UserMapperTest {
/**
* insert one Tenant
+ *
* @return Tenant
*/
- private Tenant insertOneTenant(Queue queue){
+ private Tenant insertOneTenant(Queue queue) {
Tenant tenant = new Tenant();
tenant.setTenantCode("dolphin");
tenant.setTenantName("dolphin test");
@@ -198,9 +190,10 @@ public class UserMapperTest {
/**
* insert one Queue
+ *
* @return Queue
*/
- private Queue insertOneQueue(){
+ private Queue insertOneQueue() {
Queue queue = new Queue();
queue.setQueue("dolphin");
queue.setQueueName("dolphin queue");
@@ -214,7 +207,7 @@ public class UserMapperTest {
* test update
*/
@Test
- public void testUpdate(){
+ public void testUpdate() {
//insertOne
User user = insertOne();
//update
@@ -229,7 +222,7 @@ public class UserMapperTest {
* test delete
*/
@Test
- public void testDelete(){
+ public void testDelete() {
//insertOne
User user = insertOne();
//delete
@@ -261,30 +254,6 @@ public class UserMapperTest {
Assert.assertNotEquals(userList.size(), 0);
}
-// /**
-// * test query by username
-// */
-// @Test
-// public void testQueryByUserNameAccurately() {
-// //insertOne
-// User user = insertOne();
-// //queryByUserNameAccurately
-// User queryUser = userMapper.queryByUserNameAccurately(user.getUserName());
-// Assert.assertEquals(queryUser.getUserName(), user.getUserName());
-// }
-
-// /**
-// * test query by username and password
-// */
-// @Test
-// public void testQueryUserByNamePassword() {
-// //insertOne
-// User user = insertOne();
-// //queryUserByNamePassword
-// User queryUser = userMapper.queryUserByNamePassword(user.getUserName(),user.getUserPassword());
-// Assert.assertEquals(queryUser.getUserName(),user.getUserName());
-// Assert.assertEquals(queryUser.getUserPassword(), user.getUserPassword());
-// }
/**
* test page
@@ -296,9 +265,9 @@ public class UserMapperTest {
//insertOneTenant
Tenant tenant = insertOneTenant();
//insertOne
- User user = insertOne(queue,tenant);
+ User user = insertOne(queue, tenant);
//queryUserPaging
- Page<User> page = new Page(1,3);
+ Page<User> page = new Page(1, 3);
IPage<User> userIPage = userMapper.queryUserPaging(page, user.getUserName());
Assert.assertNotEquals(userIPage.getTotal(), 0);
}
@@ -311,30 +280,13 @@ public class UserMapperTest {
//insertOneQueue and insertOneTenant
Queue queue = insertOneQueue();
Tenant tenant = insertOneTenant(queue);
- User user = insertOne(queue,tenant);
+ User user = insertOne(queue, tenant);
//queryDetailsById
User queryUser = userMapper.queryDetailsById(user.getId());
Assert.assertEquals(user.getUserName(), queryUser.getUserName());
}
/**
- * test query user list by alertgroupId
- */
- @Test
- public void testQueryUserListByAlertGroupId() {
- //insertOne
- User user = insertOne();
- //insertOneAlertGroup
- AlertGroup alertGroup = insertOneAlertGroup();
- //insertOneUserAlertGroup
- UserAlertGroup userAlertGroup = insertOneUserAlertGroup(user, alertGroup);
- //queryUserListByAlertGroupId
- List<User> userList = userMapper.queryUserListByAlertGroupId(userAlertGroup.getAlertgroupId());
- Assert.assertNotEquals(userList.size(), 0);
-
- }
-
- /**
* test query tenant code by userId
*/
@Test
@@ -345,7 +297,7 @@ public class UserMapperTest {
User user = insertOne(tenant);
//queryTenantCodeByUserId
User queryUser = userMapper.queryTenantCodeByUserId(user.getId());
- Assert.assertEquals(queryUser,user);
+ Assert.assertEquals(queryUser, user);
}
/**
@@ -359,7 +311,7 @@ public class UserMapperTest {
AccessToken accessToken = insertOneAccessToken(user);
//queryUserByToken
User userToken = userMapper.queryUserByToken(accessToken.getToken());
- Assert.assertEquals(userToken,user);
+ Assert.assertEquals(userToken, user);
}
}
diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterTaskExecThread.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterTaskExecThread.java
index 72ee0fc..b55ecbb 100644
--- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterTaskExecThread.java
+++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterTaskExecThread.java
@@ -14,8 +14,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.dolphinscheduler.server.master.runner;
+package org.apache.dolphinscheduler.server.master.runner;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
@@ -42,7 +42,6 @@ import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import java.util.Date;
import java.util.Set;
-
/**
* master task exec thread
*/
@@ -64,9 +63,10 @@ public class MasterTaskExecThread extends MasterBaseTaskExecThread {
/**
* constructor of MasterTaskExecThread
- * @param taskInstance task instance
+ *
+ * @param taskInstance task instance
*/
- public MasterTaskExecThread(TaskInstance taskInstance){
+ public MasterTaskExecThread(TaskInstance taskInstance) {
super(taskInstance);
this.taskInstanceCacheManager = SpringApplicationContext.getBean(TaskInstanceCacheManagerImpl.class);
this.nettyExecutorManager = SpringApplicationContext.getBean(NettyExecutorManager.class);
@@ -75,10 +75,11 @@ public class MasterTaskExecThread extends MasterBaseTaskExecThread {
/**
* get task instance
+ *
* @return TaskInstance
*/
@Override
- public TaskInstance getTaskInstance(){
+ public TaskInstance getTaskInstance() {
return this.taskInstance;
}
@@ -96,70 +97,70 @@ public class MasterTaskExecThread extends MasterBaseTaskExecThread {
public Boolean submitWaitComplete() {
Boolean result = false;
this.taskInstance = submit();
- if(this.taskInstance == null){
+ if (this.taskInstance == null) {
logger.error("submit task instance to mysql and queue failed , please check and fix it");
return result;
}
- if(!this.taskInstance.getState().typeIsFinished()) {
+ if (!this.taskInstance.getState().typeIsFinished()) {
result = waitTaskQuit();
}
taskInstance.setEndTime(new Date());
processService.updateTaskInstance(taskInstance);
logger.info("task :{} id:{}, process id:{}, exec thread completed ",
- this.taskInstance.getName(),taskInstance.getId(), processInstance.getId() );
+ this.taskInstance.getName(), taskInstance.getId(), processInstance.getId());
return result;
}
/**
* polling db
- *
+ * <p>
* wait task quit
+ *
* @return true if task quit success
*/
- public Boolean waitTaskQuit(){
+ public Boolean waitTaskQuit() {
// query new state
taskInstance = processService.findTaskInstanceById(taskInstance.getId());
logger.info("wait task: process id: {}, task id:{}, task name:{} complete",
- this.taskInstance.getProcessInstanceId(), this.taskInstance.getId(), this.taskInstance.getName());
+ this.taskInstance.getProcessInstanceId(), this.taskInstance.getId(), this.taskInstance.getName());
// task time out
boolean checkTimeout = false;
TaskTimeoutParameter taskTimeoutParameter = getTaskTimeoutParameter();
- if(taskTimeoutParameter.getEnable()){
+ if (taskTimeoutParameter.getEnable()) {
TaskTimeoutStrategy strategy = taskTimeoutParameter.getStrategy();
- if(strategy == TaskTimeoutStrategy.WARN || strategy == TaskTimeoutStrategy.WARNFAILED){
+ if (strategy == TaskTimeoutStrategy.WARN || strategy == TaskTimeoutStrategy.WARNFAILED) {
checkTimeout = true;
}
}
- while (Stopper.isRunning()){
+ while (Stopper.isRunning()) {
try {
- if(this.processInstance == null){
+ if (this.processInstance == null) {
logger.error("process instance not exists , master task exec thread exit");
return true;
}
// task instance add queue , waiting worker to kill
- if(this.cancel || this.processInstance.getState() == ExecutionStatus.READY_STOP){
+ if (this.cancel || this.processInstance.getState() == ExecutionStatus.READY_STOP) {
cancelTaskInstance();
}
- if(processInstance.getState() == ExecutionStatus.READY_PAUSE){
+ if (processInstance.getState() == ExecutionStatus.READY_PAUSE) {
pauseTask();
}
// task instance finished
- if (taskInstance.getState().typeIsFinished()){
+ if (taskInstance.getState().typeIsFinished()) {
// if task is final result , then remove taskInstance from cache
taskInstanceCacheManager.removeByTaskInstanceId(taskInstance.getId());
break;
}
- if(checkTimeout){
+ if (checkTimeout) {
long remainTime = DateUtils.getRemainTime(taskInstance.getStartTime(), taskTimeoutParameter.getInterval() * 60L);
if (remainTime < 0) {
- logger.warn("task id: {} execution time out",taskInstance.getId());
+ logger.warn("task id: {} execution time out", taskInstance.getId());
// process define
ProcessDefinition processDefine = processService.findProcessDefineById(processInstance.getProcessDefinitionId());
// send warn mail
- alertDao.sendTaskTimeoutAlert(processInstance.getWarningGroupId(),processDefine.getReceivers(),
- processDefine.getReceiversCc(), processInstance.getId(), processInstance.getName(),
- taskInstance.getId(),taskInstance.getName());
+ alertDao.sendTaskTimeoutAlert(processInstance.getWarningGroupId(), processInstance.getId(), processInstance.getName(),
+ taskInstance.getId(), taskInstance.getName());
checkTimeout = false;
}
}
@@ -168,10 +169,10 @@ public class MasterTaskExecThread extends MasterBaseTaskExecThread {
processInstance = processService.findProcessInstanceById(processInstance.getId());
Thread.sleep(Constants.SLEEP_TIME_MILLIS);
} catch (Exception e) {
- logger.error("exception",e);
+ logger.error("exception", e);
if (processInstance != null) {
logger.error("wait task quit failed, instance id:{}, task id:{}",
- processInstance.getId(), taskInstance.getId());
+ processInstance.getId(), taskInstance.getId());
}
}
}
@@ -180,31 +181,29 @@ public class MasterTaskExecThread extends MasterBaseTaskExecThread {
/**
* pause task if task have not been dispatched to worker, do not dispatch anymore.
- *
*/
public void pauseTask() {
taskInstance = processService.findTaskInstanceById(taskInstance.getId());
- if(taskInstance == null){
+ if (taskInstance == null) {
return;
}
- if(StringUtils.isBlank(taskInstance.getHost())){
+ if (StringUtils.isBlank(taskInstance.getHost())) {
taskInstance.setState(ExecutionStatus.PAUSE);
taskInstance.setEndTime(new Date());
processService.updateTaskInstance(taskInstance);
}
}
-
/**
- * task instance add queue , waiting worker to kill
+ * task instance add queue , waiting worker to kill
*/
- private void cancelTaskInstance() throws Exception{
- if(alreadyKilled){
+ private void cancelTaskInstance() throws Exception {
+ if (alreadyKilled) {
return;
}
alreadyKilled = true;
taskInstance = processService.findTaskInstanceById(taskInstance.getId());
- if(StringUtils.isBlank(taskInstance.getHost())){
+ if (StringUtils.isBlank(taskInstance.getHost())) {
taskInstance.setState(ExecutionStatus.KILL);
taskInstance.setEndTime(new Date());
processService.updateTaskInstance(taskInstance);
@@ -222,23 +221,24 @@ public class MasterTaskExecThread extends MasterBaseTaskExecThread {
nettyExecutorManager.executeDirectly(executionContext);
logger.info("master kill taskInstance name :{} taskInstance id:{}",
- taskInstance.getName(), taskInstance.getId() );
+ taskInstance.getName(), taskInstance.getId());
}
/**
* whether exists valid worker group
+ *
* @param taskInstanceWorkerGroup taskInstanceWorkerGroup
* @return whether exists
*/
- public Boolean existsValidWorkerGroup(String taskInstanceWorkerGroup){
+ public Boolean existsValidWorkerGroup(String taskInstanceWorkerGroup) {
Set<String> workerGroups = zookeeperRegistryCenter.getWorkerGroupDirectly();
// not worker group
- if (CollectionUtils.isEmpty(workerGroups)){
+ if (CollectionUtils.isEmpty(workerGroups)) {
return false;
}
// has worker group , but not taskInstance assigned worker group
- if (!workerGroups.contains(taskInstanceWorkerGroup)){
+ if (!workerGroups.contains(taskInstanceWorkerGroup)) {
return false;
}
Set<String> workers = zookeeperRegistryCenter.getWorkerGroupNodesDirectly(taskInstanceWorkerGroup);
@@ -250,9 +250,10 @@ public class MasterTaskExecThread extends MasterBaseTaskExecThread {
/**
* get task timeout parameter
+ *
* @return TaskTimeoutParameter
*/
- private TaskTimeoutParameter getTaskTimeoutParameter(){
+ private TaskTimeoutParameter getTaskTimeoutParameter() {
String taskJson = taskInstance.getTaskJson();
TaskNode taskNode = JSONUtils.parseObject(taskJson, TaskNode.class);
return taskNode.getTaskTimeoutParameter();
diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/AlertManager.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/AlertManager.java
index 073c075..f1d351d 100644
--- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/AlertManager.java
+++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/AlertManager.java
@@ -17,7 +17,6 @@
package org.apache.dolphinscheduler.server.utils;
-import org.apache.dolphinscheduler.common.enums.AlertType;
import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
@@ -28,7 +27,6 @@ import org.apache.dolphinscheduler.dao.entity.ProcessAlertContent;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
-import org.apache.dolphinscheduler.spi.alert.ShowType;
import java.util.ArrayList;
import java.util.Date;
@@ -171,14 +169,10 @@ public class AlertManager {
try {
Alert alert = new Alert();
alert.setTitle("worker fault tolerance");
- //alert.setShowType(ShowType.TABLE);
String content = getWorkerToleranceContent(processInstance, toleranceTaskList);
alert.setContent(content);
- alert.setAlertType(AlertType.EMAIL);
alert.setCreateTime(new Date());
alert.setAlertGroupId(processInstance.getWarningGroupId() == null ? 1 : processInstance.getWarningGroupId());
- alert.setReceivers(processInstance.getProcessDefinition().getReceivers());
- alert.setReceiversCc(processInstance.getProcessDefinition().getReceiversCc());
alertDao.addAlert(alert);
logger.info("add alert to db , alert : {}", alert.toString());
@@ -225,16 +219,10 @@ public class AlertManager {
String cmdName = getCommandCnName(processInstance.getCommandType());
String success = processInstance.getState().typeIsSuccess() ? "success" : "failed";
alert.setTitle(cmdName + " " + success);
- ShowType showType = processInstance.getState().typeIsSuccess() ? ShowType.TEXT : ShowType.TABLE;
- //alert.setShowType(showType);
String content = getContentProcessInstance(processInstance, taskInstances);
alert.setContent(content);
- alert.setAlertType(AlertType.EMAIL);
alert.setAlertGroupId(processInstance.getWarningGroupId());
alert.setCreateTime(new Date());
- alert.setReceivers(processInstance.getProcessDefinition().getReceivers());
- alert.setReceiversCc(processInstance.getProcessDefinition().getReceiversCc());
-
alertDao.addAlert(alert);
logger.info("add alert to db , alert: {}", alert.toString());
}
diff --git a/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/registry/DependencyConfig.java b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/registry/DependencyConfig.java
index 4d7b50f..d51d64d 100644
--- a/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/registry/DependencyConfig.java
+++ b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/registry/DependencyConfig.java
@@ -34,7 +34,6 @@ import org.apache.dolphinscheduler.dao.mapper.ScheduleMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.dao.mapper.UdfFuncMapper;
-import org.apache.dolphinscheduler.dao.mapper.UserAlertGroupMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.dolphinscheduler.server.master.cache.impl.TaskInstanceCacheManagerImpl;
import org.apache.dolphinscheduler.server.master.config.MasterConfig;
@@ -67,11 +66,6 @@ public class DependencyConfig {
}
@Bean
- public UserAlertGroupMapper userAlertGroupMapper() {
- return Mockito.mock(UserAlertGroupMapper.class);
- }
-
- @Bean
public TaskInstanceCacheManagerImpl taskInstanceCacheManagerImpl() {
return Mockito.mock(TaskInstanceCacheManagerImpl.class);
}
@@ -136,7 +130,6 @@ public class DependencyConfig {
return Mockito.mock(ResourceMapper.class);
}
-
@Bean
public ErrorCommandMapper errorCommandMapper() {
return Mockito.mock(ErrorCommandMapper.class);
diff --git a/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackServiceTestConfig.java b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackServiceTestConfig.java
index 8bdc07b..224764c 100644
--- a/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackServiceTestConfig.java
+++ b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackServiceTestConfig.java
@@ -34,7 +34,6 @@ import org.apache.dolphinscheduler.dao.mapper.ScheduleMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.dao.mapper.UdfFuncMapper;
-import org.apache.dolphinscheduler.dao.mapper.UserAlertGroupMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.dolphinscheduler.server.master.cache.impl.TaskInstanceCacheManagerImpl;
import org.apache.dolphinscheduler.service.process.ProcessService;
@@ -49,7 +48,6 @@ import org.springframework.context.annotation.Configuration;
@Configuration
public class TaskCallbackServiceTestConfig {
-
@Bean
public AlertDao alertDao() {
return new AlertDao();
@@ -61,11 +59,6 @@ public class TaskCallbackServiceTestConfig {
}
@Bean
- public UserAlertGroupMapper userAlertGroupMapper() {
- return Mockito.mock(UserAlertGroupMapper.class);
- }
-
- @Bean
public TaskInstanceCacheManagerImpl taskInstanceCacheManagerImpl() {
return Mockito.mock(TaskInstanceCacheManagerImpl.class);
}
@@ -125,7 +118,6 @@ public class TaskCallbackServiceTestConfig {
return Mockito.mock(ResourceMapper.class);
}
-
@Bean
public ErrorCommandMapper errorCommandMapper() {
return Mockito.mock(ErrorCommandMapper.class);
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
index 7fca374..5dde6d8 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
@@ -14,6 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.service.process;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_END_DATE;
@@ -108,11 +109,11 @@ public class ProcessService {
private final Logger logger = LoggerFactory.getLogger(getClass());
- private final int[] stateArray = new int[]{ExecutionStatus.SUBMITTED_SUCCESS.ordinal(),
- ExecutionStatus.RUNNING_EXECUTION.ordinal(),
- ExecutionStatus.DELAY_EXECUTION.ordinal(),
- ExecutionStatus.READY_PAUSE.ordinal(),
- ExecutionStatus.READY_STOP.ordinal()};
+ private final int[] stateArray = new int[] {ExecutionStatus.SUBMITTED_SUCCESS.ordinal(),
+ ExecutionStatus.RUNNING_EXECUTION.ordinal(),
+ ExecutionStatus.DELAY_EXECUTION.ordinal(),
+ ExecutionStatus.READY_PAUSE.ordinal(),
+ ExecutionStatus.READY_STOP.ordinal()};
@Autowired
private UserMapper userMapper;
@@ -145,7 +146,6 @@ public class ProcessService {
private ResourceMapper resourceMapper;
-
@Autowired
private ErrorCommandMapper errorCommandMapper;
@@ -153,26 +153,27 @@ public class ProcessService {
private TenantMapper tenantMapper;
@Autowired
- private ProjectMapper projectMapper;
+ private ProjectMapper projectMapper;
/**
* handle Command (construct ProcessInstance from Command) , wrapped in transaction
- * @param logger logger
- * @param host host
+ *
+ * @param logger logger
+ * @param host host
* @param validThreadNum validThreadNum
- * @param command found command
+ * @param command found command
* @return process instance
*/
@Transactional(rollbackFor = RuntimeException.class)
public ProcessInstance handleCommand(Logger logger, String host, int validThreadNum, Command command) {
ProcessInstance processInstance = constructProcessInstance(command, host);
//cannot construct process instance, return null;
- if(processInstance == null){
+ if (processInstance == null) {
logger.error("scan command, command parameter is error: {}", command);
moveToErrorCommand(command, "process instance is null");
return null;
}
- if(!checkThreadNum(command, validThreadNum)){
+ if (!checkThreadNum(command, validThreadNum)) {
logger.info("there is not enough thread for this command: {}", command);
return setWaitingThreadProcess(command, processInstance);
}
@@ -186,6 +187,7 @@ public class ProcessService {
/**
* save error command, and delete original command
+ *
* @param command command
* @param message message
*/
@@ -198,13 +200,14 @@ public class ProcessService {
/**
* set process waiting thread
- * @param command command
+ *
+ * @param command command
* @param processInstance processInstance
* @return process instance
*/
private ProcessInstance setWaitingThreadProcess(Command command, ProcessInstance processInstance) {
processInstance.setState(ExecutionStatus.WAITTING_THREAD);
- if(command.getCommandType() != CommandType.RECOVER_WAITTING_THREAD){
+ if (command.getCommandType() != CommandType.RECOVER_WAITTING_THREAD) {
processInstance.addHistoryCmd(command.getCommandType());
}
saveProcessInstance(processInstance);
@@ -215,7 +218,8 @@ public class ProcessService {
/**
* check thread num
- * @param command command
+ *
+ * @param command command
* @param validThreadNum validThreadNum
* @return if thread is enough
*/
@@ -226,12 +230,13 @@ public class ProcessService {
/**
* insert one command
+ *
* @param command command
* @return create result
*/
public int createCommand(Command command) {
int result = 0;
- if (command != null){
+ if (command != null) {
result = commandMapper.insert(command);
}
return result;
@@ -239,59 +244,63 @@ public class ProcessService {
/**
* find one command from queue list
+ *
* @return command
*/
- public Command findOneCommand(){
+ public Command findOneCommand() {
return commandMapper.getOneToRun();
}
/**
* check the input command exists in queue list
+ *
* @param command command
* @return create command result
*/
- public Boolean verifyIsNeedCreateCommand(Command command){
+ public Boolean verifyIsNeedCreateCommand(Command command) {
Boolean isNeedCreate = true;
- Map<CommandType,Integer> cmdTypeMap = new HashMap<CommandType,Integer>();
- cmdTypeMap.put(CommandType.REPEAT_RUNNING,1);
- cmdTypeMap.put(CommandType.RECOVER_SUSPENDED_PROCESS,1);
- cmdTypeMap.put(CommandType.START_FAILURE_TASK_PROCESS,1);
+ Map<CommandType, Integer> cmdTypeMap = new HashMap<CommandType, Integer>();
+ cmdTypeMap.put(CommandType.REPEAT_RUNNING, 1);
+ cmdTypeMap.put(CommandType.RECOVER_SUSPENDED_PROCESS, 1);
+ cmdTypeMap.put(CommandType.START_FAILURE_TASK_PROCESS, 1);
CommandType commandType = command.getCommandType();
- if(cmdTypeMap.containsKey(commandType)){
+ if (cmdTypeMap.containsKey(commandType)) {
ObjectNode cmdParamObj = JSONUtils.parseObject(command.getCommandParam());
int processInstanceId = cmdParamObj.path(CMDPARAM_RECOVER_PROCESS_ID_STRING).asInt();
List<Command> commands = commandMapper.selectList(null);
// for all commands
- for (Command tmpCommand:commands){
- if(cmdTypeMap.containsKey(tmpCommand.getCommandType())){
+ for (Command tmpCommand : commands) {
+ if (cmdTypeMap.containsKey(tmpCommand.getCommandType())) {
ObjectNode tempObj = JSONUtils.parseObject(tmpCommand.getCommandParam());
- if(tempObj != null && processInstanceId == tempObj.path(CMDPARAM_RECOVER_PROCESS_ID_STRING).asInt()){
+ if (tempObj != null && processInstanceId == tempObj.path(CMDPARAM_RECOVER_PROCESS_ID_STRING).asInt()) {
isNeedCreate = false;
break;
}
}
}
}
- return isNeedCreate;
+ return isNeedCreate;
}
/**
* find process instance detail by id
+ *
* @param processId processId
* @return process instance
*/
- public ProcessInstance findProcessInstanceDetailById(int processId){
+ public ProcessInstance findProcessInstanceDetailById(int processId) {
return processInstanceMapper.queryDetailById(processId);
}
/**
* get task node list by definitionId
+ *
* @param defineId
* @return
*/
- public List<TaskNode> getTaskNodeListByDefinitionId(Integer defineId){
+ public List<TaskNode> getTaskNodeListByDefinitionId(Integer defineId) {
ProcessDefinition processDefinition = processDefineMapper.selectById(defineId);
if (processDefinition == null) {
logger.info("process define not exists");
@@ -312,15 +321,17 @@ public class ProcessService {
/**
* find process instance by id
+ *
* @param processId processId
* @return process instance
*/
- public ProcessInstance findProcessInstanceById(int processId){
+ public ProcessInstance findProcessInstanceById(int processId) {
return processInstanceMapper.selectById(processId);
}
/**
* find process define by id.
+ *
* @param processDefinitionId processDefinitionId
* @return process definition
*/
@@ -330,23 +341,25 @@ public class ProcessService {
/**
* delete work process instance by id
+ *
* @param processInstanceId processInstanceId
* @return delete process instance result
*/
- public int deleteWorkProcessInstanceById(int processInstanceId){
+ public int deleteWorkProcessInstanceById(int processInstanceId) {
return processInstanceMapper.deleteById(processInstanceId);
}
/**
* delete all sub process by parent instance id
+ *
* @param processInstanceId processInstanceId
* @return delete all sub process instance result
*/
- public int deleteAllSubWorkProcessByParentId(int processInstanceId){
+ public int deleteAllSubWorkProcessByParentId(int processInstanceId) {
List<Integer> subProcessIdList = processInstanceMapMapper.querySubIdListByParentId(processInstanceId);
- for(Integer subId : subProcessIdList){
+ for (Integer subId : subProcessIdList) {
deleteAllSubWorkProcessByParentId(subId);
deleteWorkProcessMapByParentId(subId);
removeTaskLogFile(subId);
@@ -355,59 +368,58 @@ public class ProcessService {
return 1;
}
-
/**
* remove task log file
+ *
* @param processInstanceId processInstanceId
*/
- public void removeTaskLogFile(Integer processInstanceId){
+ public void removeTaskLogFile(Integer processInstanceId) {
LogClientService logClient = new LogClientService();
List<TaskInstance> taskInstanceList = findValidTaskListByProcessId(processInstanceId);
- if (CollectionUtils.isEmpty(taskInstanceList)){
+ if (CollectionUtils.isEmpty(taskInstanceList)) {
return;
}
- for (TaskInstance taskInstance : taskInstanceList){
+ for (TaskInstance taskInstance : taskInstanceList) {
String taskLogPath = taskInstance.getLogPath();
- if (StringUtils.isEmpty(taskInstance.getHost())){
+ if (StringUtils.isEmpty(taskInstance.getHost())) {
continue;
}
int port = Constants.RPC_PORT;
String ip = "";
try {
ip = Host.of(taskInstance.getHost()).getIp();
- }catch (Exception e){
+ } catch (Exception e) {
// compatible old version
ip = taskInstance.getHost();
}
-
-
// remove task log from loggerserver
- logClient.removeTaskLog(ip,port,taskLogPath);
+ logClient.removeTaskLog(ip, port, taskLogPath);
}
}
-
/**
* calculate sub process number in the process define.
+ *
* @param processDefinitionId processDefinitionId
* @return process thread num count
*/
- private Integer workProcessThreadNumCount(Integer processDefinitionId){
+ private Integer workProcessThreadNumCount(Integer processDefinitionId) {
List<Integer> ids = new ArrayList<>();
recurseFindSubProcessId(processDefinitionId, ids);
- return ids.size()+1;
+ return ids.size() + 1;
}
/**
* recursive query sub process definition id by parent id.
+ *
* @param parentId parentId
- * @param ids ids
+ * @param ids ids
*/
- public void recurseFindSubProcessId(int parentId, List<Integer> ids){
+ public void recurseFindSubProcessId(int parentId, List<Integer> ids) {
ProcessDefinition processDefinition = processDefineMapper.selectById(parentId);
String processDefinitionJson = processDefinition.getProcessDefinitionJson();
@@ -415,15 +427,15 @@ public class ProcessService {
List<TaskNode> taskNodeList = processData.getTasks();
- if (taskNodeList != null && taskNodeList.size() > 0){
+ if (taskNodeList != null && taskNodeList.size() > 0) {
- for (TaskNode taskNode : taskNodeList){
+ for (TaskNode taskNode : taskNodeList) {
String parameter = taskNode.getParams();
ObjectNode parameterJson = JSONUtils.parseObject(parameter);
- if (parameterJson.get(CMDPARAM_SUB_PROCESS_DEFINE_ID) != null){
+ if (parameterJson.get(CMDPARAM_SUB_PROCESS_DEFINE_ID) != null) {
SubProcessParameters subProcessParam = JSONUtils.parseObject(parameter, SubProcessParameters.class);
ids.add(subProcessParam.getProcessDefinitionId());
- recurseFindSubProcessId(subProcessParam.getProcessDefinitionId(),ids);
+ recurseFindSubProcessId(subProcessParam.getProcessDefinitionId(), ids);
}
}
@@ -435,14 +447,15 @@ public class ProcessService {
* sub work process instance need not to create recovery command.
* create recovery waiting thread command and delete origin command at the same time.
* if the recovery command is exists, only update the field update_time
- * @param originCommand originCommand
+ *
+ * @param originCommand originCommand
* @param processInstance processInstance
*/
public void createRecoveryWaitingThreadCommand(Command originCommand, ProcessInstance processInstance) {
// sub process doesnot need to create wait command
- if(processInstance.getIsSubProcess() == Flag.YES){
- if(originCommand != null){
+ if (processInstance.getIsSubProcess() == Flag.YES) {
+ if (originCommand != null) {
commandMapper.deleteById(originCommand.getId());
}
return;
@@ -450,28 +463,28 @@ public class ProcessService {
Map<String, String> cmdParam = new HashMap<>();
cmdParam.put(Constants.CMDPARAM_RECOVERY_WAITTING_THREAD, String.valueOf(processInstance.getId()));
// process instance quit by "waiting thread" state
- if(originCommand == null){
+ if (originCommand == null) {
Command command = new Command(
- CommandType.RECOVER_WAITTING_THREAD,
- processInstance.getTaskDependType(),
- processInstance.getFailureStrategy(),
- processInstance.getExecutorId(),
- processInstance.getProcessDefinitionId(),
- JSONUtils.toJsonString(cmdParam),
- processInstance.getWarningType(),
- processInstance.getWarningGroupId(),
- processInstance.getScheduleTime(),
- processInstance.getProcessInstancePriority()
+ CommandType.RECOVER_WAITTING_THREAD,
+ processInstance.getTaskDependType(),
+ processInstance.getFailureStrategy(),
+ processInstance.getExecutorId(),
+ processInstance.getProcessDefinitionId(),
+ JSONUtils.toJsonString(cmdParam),
+ processInstance.getWarningType(),
+ processInstance.getWarningGroupId(),
+ processInstance.getScheduleTime(),
+ processInstance.getProcessInstancePriority()
);
saveCommand(command);
- return ;
+ return;
}
// update the command time if current command if recover from waiting
- if(originCommand.getCommandType() == CommandType.RECOVER_WAITTING_THREAD){
+ if (originCommand.getCommandType() == CommandType.RECOVER_WAITTING_THREAD) {
originCommand.setUpdateTime(new Date());
saveCommand(originCommand);
- }else{
+ } else {
// delete old command and create new waiting thread command
commandMapper.deleteById(originCommand.getId());
originCommand.setId(0);
@@ -485,14 +498,15 @@ public class ProcessService {
/**
* get schedule time from command
- * @param command command
+ *
+ * @param command command
* @param cmdParam cmdParam map
* @return date
*/
- private Date getScheduleTime(Command command, Map<String, String> cmdParam){
+ private Date getScheduleTime(Command command, Map<String, String> cmdParam) {
Date scheduleTime = command.getScheduleTime();
- if(scheduleTime == null){
- if(cmdParam != null && cmdParam.containsKey(CMDPARAM_COMPLEMENT_DATA_START_DATE)){
+ if (scheduleTime == null) {
+ if (cmdParam != null && cmdParam.containsKey(CMDPARAM_COMPLEMENT_DATA_START_DATE)) {
scheduleTime = DateUtils.stringToDate(cmdParam.get(CMDPARAM_COMPLEMENT_DATA_START_DATE));
}
}
@@ -501,14 +515,15 @@ public class ProcessService {
/**
* generate a new work process instance from command.
+ *
* @param processDefinition processDefinition
- * @param command command
- * @param cmdParam cmdParam map
+ * @param command command
+ * @param cmdParam cmdParam map
* @return process instance
*/
private ProcessInstance generateNewProcessInstance(ProcessDefinition processDefinition,
Command command,
- Map<String, String> cmdParam){
+ Map<String, String> cmdParam) {
ProcessInstance processInstance = new ProcessInstance(processDefinition);
processInstance.setState(ExecutionStatus.RUNNING_EXECUTION);
processInstance.setRecovery(Flag.NO);
@@ -529,7 +544,7 @@ public class ProcessService {
// schedule time
Date scheduleTime = getScheduleTime(command, cmdParam);
- if(scheduleTime != null){
+ if (scheduleTime != null) {
processInstance.setScheduleTime(scheduleTime);
}
processInstance.setCommandStartTime(command.getStartTime());
@@ -537,10 +552,10 @@ public class ProcessService {
processInstance.setConnects(processDefinition.getConnects());
// curing global params
processInstance.setGlobalParams(ParameterUtils.curingGlobalParams(
- processDefinition.getGlobalParamMap(),
- processDefinition.getGlobalParamList(),
- getCommandTypeIfComplement(processInstance, command),
- processInstance.getScheduleTime()));
+ processDefinition.getGlobalParamMap(),
+ processDefinition.getGlobalParamList(),
+ getCommandTypeIfComplement(processInstance, command),
+ processInstance.getScheduleTime()));
//copy process define json to process instance
processInstance.setProcessInstanceJson(processDefinition.getProcessDefinitionJson());
@@ -558,21 +573,22 @@ public class ProcessService {
* there is tenant id in definition, use the tenant of the definition.
* if there is not tenant id in the definiton or the tenant not exist
* use definition creator's tenant.
+ *
* @param tenantId tenantId
- * @param userId userId
+ * @param userId userId
* @return tenant
*/
- public Tenant getTenantForProcess(int tenantId, int userId){
+ public Tenant getTenantForProcess(int tenantId, int userId) {
Tenant tenant = null;
- if(tenantId >= 0){
+ if (tenantId >= 0) {
tenant = tenantMapper.queryById(tenantId);
}
- if (userId == 0){
+ if (userId == 0) {
return null;
}
- if(tenant == null){
+ if (tenant == null) {
User user = userMapper.selectById(userId);
tenant = tenantMapper.queryById(user.getTenantId());
}
@@ -581,15 +597,16 @@ public class ProcessService {
/**
* check command parameters is valid
- * @param command command
+ *
+ * @param command command
* @param cmdParam cmdParam map
* @return whether command param is valid
*/
- private Boolean checkCmdParam(Command command, Map<String, String> cmdParam){
- if(command.getTaskDependType() == TaskDependType.TASK_ONLY || command.getTaskDependType()== TaskDependType.TASK_PRE){
- if(cmdParam == null
- || !cmdParam.containsKey(Constants.CMDPARAM_START_NODE_NAMES)
- || cmdParam.get(Constants.CMDPARAM_START_NODE_NAMES).isEmpty()){
+ private Boolean checkCmdParam(Command command, Map<String, String> cmdParam) {
+ if (command.getTaskDependType() == TaskDependType.TASK_ONLY || command.getTaskDependType() == TaskDependType.TASK_PRE) {
+ if (cmdParam == null
+ || !cmdParam.containsKey(Constants.CMDPARAM_START_NODE_NAMES)
+ || cmdParam.get(Constants.CMDPARAM_START_NODE_NAMES).isEmpty()) {
logger.error("command node depend type is {}, but start nodes is null ", command.getTaskDependType());
return false;
}
@@ -599,82 +616,83 @@ public class ProcessService {
/**
* construct process instance according to one command.
+ *
* @param command command
- * @param host host
+ * @param host host
* @return process instance
*/
- private ProcessInstance constructProcessInstance(Command command, String host){
+ private ProcessInstance constructProcessInstance(Command command, String host) {
ProcessInstance processInstance = null;
CommandType commandType = command.getCommandType();
Map<String, String> cmdParam = JSONUtils.toMap(command.getCommandParam());
ProcessDefinition processDefinition = null;
- if(command.getProcessDefinitionId() != 0){
+ if (command.getProcessDefinitionId() != 0) {
processDefinition = processDefineMapper.selectById(command.getProcessDefinitionId());
- if(processDefinition == null){
+ if (processDefinition == null) {
logger.error("cannot find the work process define! define id : {}", command.getProcessDefinitionId());
return null;
}
}
- if(cmdParam != null ){
+ if (cmdParam != null) {
Integer processInstanceId = 0;
// recover from failure or pause tasks
- if(cmdParam.containsKey(Constants.CMDPARAM_RECOVER_PROCESS_ID_STRING)) {
+ if (cmdParam.containsKey(Constants.CMDPARAM_RECOVER_PROCESS_ID_STRING)) {
String processId = cmdParam.get(Constants.CMDPARAM_RECOVER_PROCESS_ID_STRING);
processInstanceId = Integer.parseInt(processId);
if (processInstanceId == 0) {
logger.error("command parameter is error, [ ProcessInstanceId ] is 0");
return null;
}
- }else if(cmdParam.containsKey(Constants.CMDPARAM_SUB_PROCESS)){
+ } else if (cmdParam.containsKey(Constants.CMDPARAM_SUB_PROCESS)) {
// sub process map
String pId = cmdParam.get(Constants.CMDPARAM_SUB_PROCESS);
processInstanceId = Integer.parseInt(pId);
- }else if(cmdParam.containsKey(Constants.CMDPARAM_RECOVERY_WAITTING_THREAD)){
+ } else if (cmdParam.containsKey(Constants.CMDPARAM_RECOVERY_WAITTING_THREAD)) {
// waiting thread command
String pId = cmdParam.get(Constants.CMDPARAM_RECOVERY_WAITTING_THREAD);
processInstanceId = Integer.parseInt(pId);
}
- if(processInstanceId ==0){
+ if (processInstanceId == 0) {
processInstance = generateNewProcessInstance(processDefinition, command, cmdParam);
- }else{
+ } else {
processInstance = this.findProcessInstanceDetailById(processInstanceId);
}
processDefinition = processDefineMapper.selectById(processInstance.getProcessDefinitionId());
processInstance.setProcessDefinition(processDefinition);
//reset command parameter
- if(processInstance.getCommandParam() != null){
+ if (processInstance.getCommandParam() != null) {
Map<String, String> processCmdParam = JSONUtils.toMap(processInstance.getCommandParam());
- for(Map.Entry<String, String> entry: processCmdParam.entrySet()) {
- if(!cmdParam.containsKey(entry.getKey())){
+ for (Map.Entry<String, String> entry : processCmdParam.entrySet()) {
+ if (!cmdParam.containsKey(entry.getKey())) {
cmdParam.put(entry.getKey(), entry.getValue());
}
}
}
// reset command parameter if sub process
- if(cmdParam.containsKey(Constants.CMDPARAM_SUB_PROCESS)){
+ if (cmdParam.containsKey(Constants.CMDPARAM_SUB_PROCESS)) {
processInstance.setCommandParam(command.getCommandParam());
}
- }else{
+ } else {
// generate one new process instance
processInstance = generateNewProcessInstance(processDefinition, command, cmdParam);
}
- if(!checkCmdParam(command, cmdParam)){
+ if (!checkCmdParam(command, cmdParam)) {
logger.error("command parameter check failed!");
return null;
}
- if(command.getScheduleTime() != null){
+ if (command.getScheduleTime() != null) {
processInstance.setScheduleTime(command.getScheduleTime());
}
processInstance.setHost(host);
ExecutionStatus runStatus = ExecutionStatus.RUNNING_EXECUTION;
int runTime = processInstance.getRunTimes();
- switch (commandType){
+ switch (commandType) {
case START_PROCESS:
break;
case START_FAILURE_TASK_PROCESS:
@@ -686,13 +704,13 @@ public class ProcessService {
failedList.addAll(killedList);
failedList.addAll(toleranceList);
- for(Integer taskId : failedList){
+ for (Integer taskId : failedList) {
initTaskInstance(this.findTaskInstanceById(taskId));
}
cmdParam.put(Constants.CMDPARAM_RECOVERY_START_NODE_STRING,
- String.join(Constants.COMMA, convertIntListToString(failedList)));
+ String.join(Constants.COMMA, convertIntListToString(failedList)));
processInstance.setCommandParam(JSONUtils.toJsonString(cmdParam));
- processInstance.setRunTimes(runTime +1 );
+ processInstance.setRunTimes(runTime + 1);
break;
case START_CURRENT_TASK_PROCESS:
break;
@@ -703,15 +721,15 @@ public class ProcessService {
cmdParam.remove(Constants.CMDPARAM_RECOVERY_START_NODE_STRING);
List<Integer> suspendedNodeList = this.findTaskIdByInstanceState(processInstance.getId(), ExecutionStatus.PAUSE);
List<Integer> stopNodeList = findTaskIdByInstanceState(processInstance.getId(),
- ExecutionStatus.KILL);
+ ExecutionStatus.KILL);
suspendedNodeList.addAll(stopNodeList);
- for(Integer taskId : suspendedNodeList){
+ for (Integer taskId : suspendedNodeList) {
// initialize the pause state
initTaskInstance(this.findTaskInstanceById(taskId));
}
cmdParam.put(Constants.CMDPARAM_RECOVERY_START_NODE_STRING, String.join(",", convertIntListToString(suspendedNodeList)));
processInstance.setCommandParam(JSONUtils.toJsonString(cmdParam));
- processInstance.setRunTimes(runTime +1);
+ processInstance.setRunTimes(runTime + 1);
break;
case RECOVER_TOLERANCE_FAULT_PROCESS:
// recover tolerance fault process
@@ -721,7 +739,7 @@ public class ProcessService {
case COMPLEMENT_DATA:
// delete all the valid tasks when complement data
List<TaskInstance> taskInstanceList = this.findValidTaskListByProcessId(processInstance.getId());
- for(TaskInstance taskInstance : taskInstanceList){
+ for (TaskInstance taskInstance : taskInstanceList) {
taskInstance.setFlag(Flag.NO);
this.updateTaskInstance(taskInstance);
}
@@ -729,19 +747,19 @@ public class ProcessService {
break;
case REPEAT_RUNNING:
// delete the recover task names from command parameter
- if(cmdParam.containsKey(Constants.CMDPARAM_RECOVERY_START_NODE_STRING)){
+ if (cmdParam.containsKey(Constants.CMDPARAM_RECOVERY_START_NODE_STRING)) {
cmdParam.remove(Constants.CMDPARAM_RECOVERY_START_NODE_STRING);
processInstance.setCommandParam(JSONUtils.toJsonString(cmdParam));
}
// delete all the valid tasks when repeat running
List<TaskInstance> validTaskList = findValidTaskListByProcessId(processInstance.getId());
- for(TaskInstance taskInstance : validTaskList){
+ for (TaskInstance taskInstance : validTaskList) {
taskInstance.setFlag(Flag.NO);
updateTaskInstance(taskInstance);
}
processInstance.setStartTime(new Date());
processInstance.setEndTime(null);
- processInstance.setRunTimes(runTime +1);
+ processInstance.setRunTimes(runTime + 1);
initComplementDataParam(processDefinition, processInstance, cmdParam);
break;
case SCHEDULER:
@@ -755,60 +773,62 @@ public class ProcessService {
/**
* return complement data if the process start with complement data
+ *
* @param processInstance processInstance
- * @param command command
+ * @param command command
* @return command type
*/
- private CommandType getCommandTypeIfComplement(ProcessInstance processInstance, Command command){
- if(CommandType.COMPLEMENT_DATA == processInstance.getCmdTypeIfComplement()){
+ private CommandType getCommandTypeIfComplement(ProcessInstance processInstance, Command command) {
+ if (CommandType.COMPLEMENT_DATA == processInstance.getCmdTypeIfComplement()) {
return CommandType.COMPLEMENT_DATA;
- }else{
+ } else {
return command.getCommandType();
}
}
/**
* initialize complement data parameters
+ *
* @param processDefinition processDefinition
- * @param processInstance processInstance
- * @param cmdParam cmdParam
+ * @param processInstance processInstance
+ * @param cmdParam cmdParam
*/
private void initComplementDataParam(ProcessDefinition processDefinition,
ProcessInstance processInstance,
Map<String, String> cmdParam) {
- if(!processInstance.isComplementData()){
+ if (!processInstance.isComplementData()) {
return;
}
Date startComplementTime = DateUtils.parse(cmdParam.get(CMDPARAM_COMPLEMENT_DATA_START_DATE),
- YYYY_MM_DD_HH_MM_SS);
- if(Flag.NO == processInstance.getIsSubProcess()) {
+ YYYY_MM_DD_HH_MM_SS);
+ if (Flag.NO == processInstance.getIsSubProcess()) {
processInstance.setScheduleTime(startComplementTime);
}
processInstance.setGlobalParams(ParameterUtils.curingGlobalParams(
- processDefinition.getGlobalParamMap(),
- processDefinition.getGlobalParamList(),
- CommandType.COMPLEMENT_DATA, processInstance.getScheduleTime()));
+ processDefinition.getGlobalParamMap(),
+ processDefinition.getGlobalParamList(),
+ CommandType.COMPLEMENT_DATA, processInstance.getScheduleTime()));
}
-
/**
* set sub work process parameters.
* handle sub work process instance, update relation table and command parameters
* set sub work process flag, extends parent work process command parameters
+ *
* @param subProcessInstance subProcessInstance
* @return process instance
*/
- public ProcessInstance setSubProcessParam(ProcessInstance subProcessInstance){
+ public ProcessInstance setSubProcessParam(ProcessInstance subProcessInstance) {
String cmdParam = subProcessInstance.getCommandParam();
- if(StringUtils.isEmpty(cmdParam)){
+ if (StringUtils.isEmpty(cmdParam)) {
return subProcessInstance;
}
Map<String, String> paramMap = JSONUtils.toMap(cmdParam);
// write sub process id into cmd param.
- if(paramMap.containsKey(CMDPARAM_SUB_PROCESS)
- && CMDPARAM_EMPTY_SUB_PROCESS.equals(paramMap.get(CMDPARAM_SUB_PROCESS))){
+ if (paramMap.containsKey(CMDPARAM_SUB_PROCESS)
+ && CMDPARAM_EMPTY_SUB_PROCESS.equals(paramMap.get(CMDPARAM_SUB_PROCESS))) {
paramMap.remove(CMDPARAM_SUB_PROCESS);
paramMap.put(CMDPARAM_SUB_PROCESS, String.valueOf(subProcessInstance.getId()));
subProcessInstance.setCommandParam(JSONUtils.toJsonString(paramMap));
@@ -817,18 +837,18 @@ public class ProcessService {
}
// copy parent instance user def params to sub process..
String parentInstanceId = paramMap.get(CMDPARAM_SUB_PROCESS_PARENT_INSTANCE_ID);
- if(StringUtils.isNotEmpty(parentInstanceId)){
+ if (StringUtils.isNotEmpty(parentInstanceId)) {
ProcessInstance parentInstance = findProcessInstanceDetailById(Integer.parseInt(parentInstanceId));
- if(parentInstance != null){
+ if (parentInstance != null) {
subProcessInstance.setGlobalParams(
- joinGlobalParams(parentInstance.getGlobalParams(), subProcessInstance.getGlobalParams()));
+ joinGlobalParams(parentInstance.getGlobalParams(), subProcessInstance.getGlobalParams()));
this.saveProcessInstance(subProcessInstance);
- }else{
+ } else {
logger.error("sub process command params error, cannot find parent instance: {} ", cmdParam);
}
}
ProcessInstanceMap processInstanceMap = JSONUtils.parseObject(cmdParam, ProcessInstanceMap.class);
- if(processInstanceMap == null || processInstanceMap.getParentProcessInstanceId() == 0){
+ if (processInstanceMap == null || processInstanceMap.getParentProcessInstanceId() == 0) {
return subProcessInstance;
}
// update sub process id to process map table
@@ -841,19 +861,20 @@ public class ProcessService {
/**
* join parent global params into sub process.
* only the keys doesn't in sub process global would be joined.
+ *
* @param parentGlobalParams parentGlobalParams
- * @param subGlobalParams subGlobalParams
+ * @param subGlobalParams subGlobalParams
* @return global params join
*/
- private String joinGlobalParams(String parentGlobalParams, String subGlobalParams){
+ private String joinGlobalParams(String parentGlobalParams, String subGlobalParams) {
List<Property> parentPropertyList = JSONUtils.toList(parentGlobalParams, Property.class);
List<Property> subPropertyList = JSONUtils.toList(subGlobalParams, Property.class);
- Map<String,String> subMap = subPropertyList.stream().collect(Collectors.toMap(Property::getProp, Property::getValue));
+ Map<String, String> subMap = subPropertyList.stream().collect(Collectors.toMap(Property::getProp, Property::getValue));
- for(Property parent : parentPropertyList){
- if(!subMap.containsKey(parent.getProp())){
+ for (Property parent : parentPropertyList) {
+ if (!subMap.containsKey(parent.getProp())) {
subPropertyList.add(parent);
}
}
@@ -862,12 +883,13 @@ public class ProcessService {
/**
* initialize task instance
+ *
* @param taskInstance taskInstance
*/
- private void initTaskInstance(TaskInstance taskInstance){
+ private void initTaskInstance(TaskInstance taskInstance) {
- if(!taskInstance.isSubProcess()){
- if(taskInstance.getState().typeIsCancel() || taskInstance.getState().typeIsFailure()){
+ if (!taskInstance.isSubProcess()) {
+ if (taskInstance.getState().typeIsCancel() || taskInstance.getState().typeIsFailure()) {
taskInstance.setFlag(Flag.NO);
updateTaskInstance(taskInstance);
return;
@@ -880,27 +902,28 @@ public class ProcessService {
/**
* submit task to db
* submit sub process to command
+ *
* @param taskInstance taskInstance
* @return task instance
*/
@Transactional(rollbackFor = RuntimeException.class)
- public TaskInstance submitTask(TaskInstance taskInstance){
+ public TaskInstance submitTask(TaskInstance taskInstance) {
ProcessInstance processInstance = this.findProcessInstanceDetailById(taskInstance.getProcessInstanceId());
logger.info("start submit task : {}, instance id:{}, state: {}",
- taskInstance.getName(), taskInstance.getProcessInstanceId(), processInstance.getState());
+ taskInstance.getName(), taskInstance.getProcessInstanceId(), processInstance.getState());
//submit to db
TaskInstance task = submitTaskInstanceToDB(taskInstance, processInstance);
- if(task == null){
+ if (task == null) {
logger.error("end submit task to db error, task name:{}, process id:{} state: {} ",
- taskInstance.getName(), taskInstance.getProcessInstance(), processInstance.getState());
+ taskInstance.getName(), taskInstance.getProcessInstance(), processInstance.getState());
return task;
}
- if(!task.getState().typeIsFinished()){
+ if (!task.getState().typeIsFinished()) {
createSubWorkProcess(processInstance, task);
}
logger.info("end submit task to db successfully:{} state:{} complete, instance id:{} state: {} ",
- taskInstance.getName(), task.getState(), processInstance.getId(), processInstance.getState());
+ taskInstance.getName(), task.getState(), processInstance.getId(), processInstance.getState());
return task;
}
@@ -909,11 +932,12 @@ public class ProcessService {
* consider o
* repeat running does not generate new sub process instance
* set map {parent instance id, task instance id, 0(child instance id)}
+ *
* @param parentInstance parentInstance
- * @param parentTask parentTask
+ * @param parentTask parentTask
* @return process instance map
*/
- private ProcessInstanceMap setProcessInstanceMap(ProcessInstance parentInstance, TaskInstance parentTask){
+ private ProcessInstanceMap setProcessInstanceMap(ProcessInstance parentInstance, TaskInstance parentTask) {
ProcessInstanceMap processMap = findWorkProcessMapByParent(parentInstance.getId(), parentTask.getId());
if (processMap != null) {
return processMap;
@@ -937,8 +961,9 @@ public class ProcessService {
/**
* find previous task work process map.
+ *
* @param parentProcessInstance parentProcessInstance
- * @param parentTask parentTask
+ * @param parentTask parentTask
* @return process instance map
*/
private ProcessInstanceMap findPreviousTaskProcessMap(ProcessInstance parentProcessInstance,
@@ -956,7 +981,7 @@ public class ProcessService {
}
}
logger.info("sub process instance is not found,parent task:{},parent instance:{}",
- parentTask.getId(), parentProcessInstance.getId());
+ parentTask.getId(), parentProcessInstance.getId());
return null;
}
@@ -990,6 +1015,7 @@ public class ProcessService {
/**
* complement data needs transform parent parameter to child.
+ *
* @param instanceMap
* @param parentProcessInstance
* @return
@@ -1011,6 +1037,7 @@ public class ProcessService {
/**
* create sub work process command
+ *
* @param parentProcessInstance
* @param childInstance
* @param instanceMap
@@ -1027,22 +1054,23 @@ public class ProcessService {
String processParam = getSubWorkFlowParam(instanceMap, parentProcessInstance);
return new Command(
- commandType,
- TaskDependType.TASK_POST,
- parentProcessInstance.getFailureStrategy(),
- parentProcessInstance.getExecutorId(),
- childDefineId,
- processParam,
- parentProcessInstance.getWarningType(),
- parentProcessInstance.getWarningGroupId(),
- parentProcessInstance.getScheduleTime(),
- parentProcessInstance.getProcessInstancePriority()
+ commandType,
+ TaskDependType.TASK_POST,
+ parentProcessInstance.getFailureStrategy(),
+ parentProcessInstance.getExecutorId(),
+ childDefineId,
+ processParam,
+ parentProcessInstance.getWarningType(),
+ parentProcessInstance.getWarningGroupId(),
+ parentProcessInstance.getScheduleTime(),
+ parentProcessInstance.getProcessInstancePriority()
);
}
/**
* initialize sub work flow state
* child instance state would be initialized when 'recovery from pause/stop/failure'
+ *
* @param childInstance
*/
private void initSubInstanceState(ProcessInstance childInstance) {
@@ -1070,42 +1098,43 @@ public class ProcessService {
}
/**
- * update sub process definition
+ * update sub process definition todo
+ *
* @param parentProcessInstance parentProcessInstance
- * @param childDefinitionId childDefinitionId
+ * @param childDefinitionId childDefinitionId
*/
private void updateSubProcessDefinitionByParent(ProcessInstance parentProcessInstance, int childDefinitionId) {
ProcessDefinition fatherDefinition = this.findProcessDefineById(parentProcessInstance.getProcessDefinitionId());
ProcessDefinition childDefinition = this.findProcessDefineById(childDefinitionId);
- if(childDefinition != null && fatherDefinition != null){
- childDefinition.setReceivers(fatherDefinition.getReceivers());
- childDefinition.setReceiversCc(fatherDefinition.getReceiversCc());
+ if (childDefinition != null && fatherDefinition != null) {
+ childDefinition.setWarningGroupId(fatherDefinition.getWarningGroupId());
processDefineMapper.updateById(childDefinition);
}
}
/**
* submit task to mysql
- * @param taskInstance taskInstance
+ *
+ * @param taskInstance taskInstance
* @param processInstance processInstance
* @return task instance
*/
- public TaskInstance submitTaskInstanceToDB(TaskInstance taskInstance, ProcessInstance processInstance){
+ public TaskInstance submitTaskInstanceToDB(TaskInstance taskInstance, ProcessInstance processInstance) {
ExecutionStatus processInstanceState = processInstance.getState();
- if(taskInstance.getState().typeIsFailure()){
- if(taskInstance.isSubProcess()){
- taskInstance.setRetryTimes(taskInstance.getRetryTimes() + 1 );
- }else {
+ if (taskInstance.getState().typeIsFailure()) {
+ if (taskInstance.isSubProcess()) {
+ taskInstance.setRetryTimes(taskInstance.getRetryTimes() + 1);
+ } else {
- if( processInstanceState != ExecutionStatus.READY_STOP
- && processInstanceState != ExecutionStatus.READY_PAUSE){
+ if (processInstanceState != ExecutionStatus.READY_STOP
+ && processInstanceState != ExecutionStatus.READY_PAUSE) {
// failure task set invalid
taskInstance.setFlag(Flag.NO);
updateTaskInstance(taskInstance);
// crate new task instance
- if(taskInstance.getState() != ExecutionStatus.NEED_FAULT_TOLERANCE){
- taskInstance.setRetryTimes(taskInstance.getRetryTimes() + 1 );
+ if (taskInstance.getState() != ExecutionStatus.NEED_FAULT_TOLERANCE) {
+ taskInstance.setRetryTimes(taskInstance.getRetryTimes() + 1);
}
taskInstance.setSubmitTime(null);
taskInstance.setStartTime(null);
@@ -1126,16 +1155,16 @@ public class ProcessService {
taskInstance.setFirstSubmitTime(taskInstance.getSubmitTime());
}
boolean saveResult = saveTaskInstance(taskInstance);
- if(!saveResult){
+ if (!saveResult) {
return null;
}
return taskInstance;
}
-
/**
* ${processInstancePriority}_${processInstanceId}_${taskInstancePriority}_${taskInstanceId}_${task executed by ip1},${ip2}...
* The tasks with the highest priority are selected by comparing the priorities of the above four levels from high to low.
+ *
* @param taskInstance taskInstance
* @return task zk queue str
*/
@@ -1143,7 +1172,7 @@ public class ProcessService {
String taskWorkerGroup = getTaskWorkerGroup(taskInstance);
ProcessInstance processInstance = this.findProcessInstanceById(taskInstance.getProcessInstanceId());
- if(processInstance == null){
+ if (processInstance == null) {
logger.error("process instance is null. please check the task info, task id: " + taskInstance.getId());
return "";
}
@@ -1151,12 +1180,12 @@ public class ProcessService {
StringBuilder sb = new StringBuilder(100);
sb.append(processInstance.getProcessInstancePriority().ordinal()).append(Constants.UNDERLINE)
- .append(taskInstance.getProcessInstanceId()).append(Constants.UNDERLINE)
- .append(taskInstance.getTaskInstancePriority().ordinal()).append(Constants.UNDERLINE)
- .append(taskInstance.getId()).append(Constants.UNDERLINE)
- .append(taskInstance.getWorkerGroup());
+ .append(taskInstance.getProcessInstanceId()).append(Constants.UNDERLINE)
+ .append(taskInstance.getTaskInstancePriority().ordinal()).append(Constants.UNDERLINE)
+ .append(taskInstance.getId()).append(Constants.UNDERLINE)
+ .append(taskInstance.getWorkerGroup());
- return sb.toString();
+ return sb.toString();
}
/**
@@ -1167,51 +1196,52 @@ public class ProcessService {
* return stop if work process state is ready stop
* if all of above are not satisfied, return submit success
*
- * @param taskInstance taskInstance
+ * @param taskInstance taskInstance
* @param processInstanceState processInstanceState
* @return process instance state
*/
- public ExecutionStatus getSubmitTaskState(TaskInstance taskInstance, ExecutionStatus processInstanceState){
+ public ExecutionStatus getSubmitTaskState(TaskInstance taskInstance, ExecutionStatus processInstanceState) {
ExecutionStatus state = taskInstance.getState();
- if(
- // running, delayed or killed
- // the task already exists in task queue
- // return state
- state == ExecutionStatus.RUNNING_EXECUTION
- || state == ExecutionStatus.DELAY_EXECUTION
- || state == ExecutionStatus.KILL
- || checkTaskExistsInTaskQueue(taskInstance)
- ){
+ if (
+ // running, delayed or killed
+ // the task already exists in task queue
+ // return state
+ state == ExecutionStatus.RUNNING_EXECUTION
+ || state == ExecutionStatus.DELAY_EXECUTION
+ || state == ExecutionStatus.KILL
+ || checkTaskExistsInTaskQueue(taskInstance)
+ ) {
return state;
}
//return pasue /stop if process instance state is ready pause / stop
// or return submit success
- if( processInstanceState == ExecutionStatus.READY_PAUSE){
+ if (processInstanceState == ExecutionStatus.READY_PAUSE) {
state = ExecutionStatus.PAUSE;
- }else if(processInstanceState == ExecutionStatus.READY_STOP
- || !checkProcessStrategy(taskInstance)) {
+ } else if (processInstanceState == ExecutionStatus.READY_STOP
+ || !checkProcessStrategy(taskInstance)) {
state = ExecutionStatus.KILL;
- }else{
+ } else {
state = ExecutionStatus.SUBMITTED_SUCCESS;
}
return state;
}
/**
- * check process instance strategy
+ * check process instance strategy
+ *
* @param taskInstance taskInstance
* @return check strategy result
*/
- private boolean checkProcessStrategy(TaskInstance taskInstance){
+ private boolean checkProcessStrategy(TaskInstance taskInstance) {
ProcessInstance processInstance = this.findProcessInstanceById(taskInstance.getProcessInstanceId());
FailureStrategy failureStrategy = processInstance.getFailureStrategy();
- if(failureStrategy == FailureStrategy.CONTINUE){
+ if (failureStrategy == FailureStrategy.CONTINUE) {
return true;
}
List<TaskInstance> taskInstances = this.findValidTaskListByProcessId(taskInstance.getProcessInstanceId());
- for(TaskInstance task : taskInstances){
- if(task.getState() == ExecutionStatus.FAILURE){
+ for (TaskInstance task : taskInstances) {
+ if (task.getState() == ExecutionStatus.FAILURE) {
return false;
}
}
@@ -1220,11 +1250,12 @@ public class ProcessService {
/**
* check the task instance existing in queue
+ *
* @param taskInstance taskInstance
* @return whether taskinstance exists queue
*/
- public boolean checkTaskExistsInTaskQueue(TaskInstance taskInstance){
- if(taskInstance.isSubProcess()){
+ public boolean checkTaskExistsInTaskQueue(TaskInstance taskInstance) {
+ if (taskInstance.isSubProcess()) {
return false;
}
@@ -1235,60 +1266,65 @@ public class ProcessService {
/**
* create a new process instance
+ *
* @param processInstance processInstance
*/
- public void createProcessInstance(ProcessInstance processInstance){
+ public void createProcessInstance(ProcessInstance processInstance) {
- if (processInstance != null){
+ if (processInstance != null) {
processInstanceMapper.insert(processInstance);
}
}
/**
* insert or update work process instance to data base
+ *
* @param processInstance processInstance
*/
- public void saveProcessInstance(ProcessInstance processInstance){
+ public void saveProcessInstance(ProcessInstance processInstance) {
- if (processInstance == null){
+ if (processInstance == null) {
logger.error("save error, process instance is null!");
- return ;
+ return;
}
- if(processInstance.getId() != 0){
+ if (processInstance.getId() != 0) {
processInstanceMapper.updateById(processInstance);
- }else{
+ } else {
createProcessInstance(processInstance);
}
}
/**
* insert or update command
+ *
* @param command command
* @return save command result
*/
- public int saveCommand(Command command){
- if(command.getId() != 0){
+ public int saveCommand(Command command) {
+ if (command.getId() != 0) {
return commandMapper.updateById(command);
- }else{
+ } else {
return commandMapper.insert(command);
}
}
/**
- * insert or update task instance
+ * insert or update task instance
+ *
* @param taskInstance taskInstance
* @return save task instance result
*/
- public boolean saveTaskInstance(TaskInstance taskInstance){
- if(taskInstance.getId() != 0){
+ public boolean saveTaskInstance(TaskInstance taskInstance) {
+ if (taskInstance.getId() != 0) {
return updateTaskInstance(taskInstance);
- }else{
+ } else {
return createTaskInstance(taskInstance);
}
}
/**
* insert task instance
+ *
* @param taskInstance taskInstance
* @return create task instance result
*/
@@ -1299,16 +1335,19 @@ public class ProcessService {
/**
* update task instance
+ *
* @param taskInstance taskInstance
* @return update task instance result
*/
- public boolean updateTaskInstance(TaskInstance taskInstance){
+ public boolean updateTaskInstance(TaskInstance taskInstance) {
int count = taskInstanceMapper.updateById(taskInstance);
return count > 0;
}
+
/**
* delete a command by id
- * @param id id
+ *
+ * @param id id
*/
public void delCommandByid(int id) {
commandMapper.deleteById(id);
@@ -1316,23 +1355,24 @@ public class ProcessService {
/**
* find task instance by id
+ *
* @param taskId task id
* @return task intance
*/
- public TaskInstance findTaskInstanceById(Integer taskId){
+ public TaskInstance findTaskInstanceById(Integer taskId) {
return taskInstanceMapper.selectById(taskId);
}
-
/**
* package task instance，associate processInstance and processDefine
+ *
* @param taskInstId taskInstId
* @return task instance
*/
- public TaskInstance getTaskInstanceDetailByTaskId(int taskInstId){
+ public TaskInstance getTaskInstanceDetailByTaskId(int taskInstId) {
// get task instance
TaskInstance taskInstance = findTaskInstanceById(taskInstId);
- if(taskInstance == null){
+ if (taskInstance == null) {
return taskInstance;
}
// get process instance
@@ -1345,88 +1385,94 @@ public class ProcessService {
return taskInstance;
}
-
/**
* get id list by task state
+ *
* @param instanceId instanceId
- * @param state state
+ * @param state state
* @return task instance states
*/
- public List<Integer> findTaskIdByInstanceState(int instanceId, ExecutionStatus state){
+ public List<Integer> findTaskIdByInstanceState(int instanceId, ExecutionStatus state) {
return taskInstanceMapper.queryTaskByProcessIdAndState(instanceId, state.ordinal());
}
/**
* find valid task list by process definition id
+ *
* @param processInstanceId processInstanceId
* @return task instance list
*/
- public List<TaskInstance> findValidTaskListByProcessId(Integer processInstanceId){
- return taskInstanceMapper.findValidTaskListByProcessId(processInstanceId, Flag.YES);
+ public List<TaskInstance> findValidTaskListByProcessId(Integer processInstanceId) {
+ return taskInstanceMapper.findValidTaskListByProcessId(processInstanceId, Flag.YES);
}
/**
* find previous task list by work process id
+ *
* @param processInstanceId processInstanceId
* @return task instance list
*/
- public List<TaskInstance> findPreviousTaskListByWorkProcessId(Integer processInstanceId){
+ public List<TaskInstance> findPreviousTaskListByWorkProcessId(Integer processInstanceId) {
return taskInstanceMapper.findValidTaskListByProcessId(processInstanceId, Flag.NO);
}
/**
* update work process instance map
+ *
* @param processInstanceMap processInstanceMap
* @return update process instance result
*/
- public int updateWorkProcessInstanceMap(ProcessInstanceMap processInstanceMap){
+ public int updateWorkProcessInstanceMap(ProcessInstanceMap processInstanceMap) {
return processInstanceMapMapper.updateById(processInstanceMap);
}
-
/**
* create work process instance map
+ *
* @param processInstanceMap processInstanceMap
* @return create process instance result
*/
- public int createWorkProcessInstanceMap(ProcessInstanceMap processInstanceMap){
+ public int createWorkProcessInstanceMap(ProcessInstanceMap processInstanceMap) {
Integer count = 0;
- if(processInstanceMap !=null){
- return processInstanceMapMapper.insert(processInstanceMap);
+ if (processInstanceMap != null) {
+ return processInstanceMapMapper.insert(processInstanceMap);
}
return count;
}
/**
* find work process map by parent process id and parent task id.
+ *
* @param parentWorkProcessId parentWorkProcessId
- * @param parentTaskId parentTaskId
+ * @param parentTaskId parentTaskId
* @return process instance map
*/
- public ProcessInstanceMap findWorkProcessMapByParent(Integer parentWorkProcessId, Integer parentTaskId){
+ public ProcessInstanceMap findWorkProcessMapByParent(Integer parentWorkProcessId, Integer parentTaskId) {
return processInstanceMapMapper.queryByParentId(parentWorkProcessId, parentTaskId);
}
/**
* delete work process map by parent process id
+ *
* @param parentWorkProcessId parentWorkProcessId
* @return delete process map result
*/
- public int deleteWorkProcessMapByParentId(int parentWorkProcessId){
+ public int deleteWorkProcessMapByParentId(int parentWorkProcessId) {
return processInstanceMapMapper.deleteByParentProcessId(parentWorkProcessId);
}
/**
* find sub process instance
+ *
* @param parentProcessId parentProcessId
- * @param parentTaskId parentTaskId
+ * @param parentTaskId parentTaskId
* @return process instance
*/
- public ProcessInstance findSubProcessInstance(Integer parentProcessId, Integer parentTaskId){
+ public ProcessInstance findSubProcessInstance(Integer parentProcessId, Integer parentTaskId) {
ProcessInstance processInstance = null;
ProcessInstanceMap processInstanceMap = processInstanceMapMapper.queryByParentId(parentProcessId, parentTaskId);
- if(processInstanceMap == null || processInstanceMap.getProcessInstanceId() == 0){
+ if (processInstanceMap == null || processInstanceMap.getProcessInstanceId() == 0) {
return processInstance;
}
processInstance = findProcessInstanceById(processInstanceMap.getProcessInstanceId());
@@ -1435,28 +1481,29 @@ public class ProcessService {
/**
* find parent process instance
+ *
* @param subProcessId subProcessId
* @return process instance
*/
public ProcessInstance findParentProcessInstance(Integer subProcessId) {
ProcessInstance processInstance = null;
ProcessInstanceMap processInstanceMap = processInstanceMapMapper.queryBySubProcessId(subProcessId);
- if(processInstanceMap == null || processInstanceMap.getProcessInstanceId() == 0){
+ if (processInstanceMap == null || processInstanceMap.getProcessInstanceId() == 0) {
return processInstance;
}
processInstance = findProcessInstanceById(processInstanceMap.getParentProcessInstanceId());
return processInstance;
}
-
/**
* change task state
- * @param state state
- * @param startTime startTime
- * @param host host
+ *
+ * @param state state
+ * @param startTime startTime
+ * @param host host
* @param executePath executePath
- * @param logPath logPath
- * @param taskInstId taskInstId
+ * @param logPath logPath
+ * @param taskInstId taskInstId
*/
public void changeTaskState(ExecutionStatus state, Date startTime, String host,
String executePath,
@@ -1473,29 +1520,31 @@ public class ProcessService {
/**
* update process instance
+ *
* @param processInstance processInstance
* @return update process instance result
*/
- public int updateProcessInstance(ProcessInstance processInstance){
+ public int updateProcessInstance(ProcessInstance processInstance) {
return processInstanceMapper.updateById(processInstance);
}
/**
* update the process instance
+ *
* @param processInstanceId processInstanceId
- * @param processJson processJson
- * @param globalParams globalParams
- * @param scheduleTime scheduleTime
- * @param flag flag
- * @param locations locations
- * @param connects connects
+ * @param processJson processJson
+ * @param globalParams globalParams
+ * @param scheduleTime scheduleTime
+ * @param flag flag
+ * @param locations locations
+ * @param connects connects
* @return update process instance result
*/
public int updateProcessInstance(Integer processInstanceId, String processJson,
String globalParams, Date scheduleTime, Flag flag,
- String locations, String connects){
+ String locations, String connects) {
ProcessInstance processInstance = processInstanceMapper.queryDetailById(processInstanceId);
- if(processInstance!= null){
+ if (processInstance != null) {
processInstance.setProcessInstanceJson(processJson);
processInstance.setGlobalParams(globalParams);
processInstance.setScheduleTime(scheduleTime);
@@ -1508,10 +1557,11 @@ public class ProcessService {
/**
* change task state
- * @param state state
- * @param endTime endTime
+ *
+ * @param state state
+ * @param endTime endTime
* @param taskInstId taskInstId
- * @param varPool varPool
+ * @param varPool varPool
*/
public void changeTaskState(ExecutionStatus state,
Date endTime,
@@ -1530,15 +1580,16 @@ public class ProcessService {
/**
* convert integer list to string list
+ *
* @param intList intList
* @return string list
*/
- public List<String> convertIntListToString(List<Integer> intList){
- if(intList == null){
+ public List<String> convertIntListToString(List<Integer> intList) {
+ if (intList == null) {
return new ArrayList<>();
}
List<String> result = new ArrayList<String>(intList.size());
- for(Integer intVar : intList){
+ for (Integer intVar : intList) {
result.add(String.valueOf(intVar));
}
return result;
@@ -1546,6 +1597,7 @@ public class ProcessService {
/**
* query schedule by id
+ *
* @param id id
* @return schedule
*/
@@ -1555,6 +1607,7 @@ public class ProcessService {
/**
* query Schedule by processDefinitionId
+ *
* @param processDefinitionId processDefinitionId
* @see Schedule
*/
@@ -1564,20 +1617,22 @@ public class ProcessService {
/**
* query need failover process instance
+ *
* @param host host
* @return process instance list
*/
- public List<ProcessInstance> queryNeedFailoverProcessInstances(String host){
+ public List<ProcessInstance> queryNeedFailoverProcessInstances(String host) {
return processInstanceMapper.queryByHostAndStatus(host, stateArray);
}
/**
* process need failover process instance
+ *
* @param processInstance processInstance
*/
@Transactional(rollbackFor = RuntimeException.class)
- public void processNeedFailoverProcessInstances(ProcessInstance processInstance){
+ public void processNeedFailoverProcessInstances(ProcessInstance processInstance) {
//1 update processInstance host is null
processInstance.setHost(Constants.NULL);
processInstanceMapper.updateById(processInstance);
@@ -1593,28 +1648,30 @@ public class ProcessService {
/**
* query all need failover task instances by host
+ *
* @param host host
* @return task instance list
*/
- public List<TaskInstance> queryNeedFailoverTaskInstances(String host){
+ public List<TaskInstance> queryNeedFailoverTaskInstances(String host) {
return taskInstanceMapper.queryByHostAndStatus(host,
- stateArray);
+ stateArray);
}
/**
* find data source by id
+ *
* @param id id
* @return datasource
*/
- public DataSource findDataSourceById(int id){
+ public DataSource findDataSourceById(int id) {
return dataSourceMapper.selectById(id);
}
-
/**
* update process instance state by id
+ *
* @param processInstanceId processInstanceId
- * @param executionStatus executionStatus
+ * @param executionStatus executionStatus
* @return update process result
*/
public int updateProcessInstanceState(Integer processInstanceId, ExecutionStatus executionStatus) {
@@ -1626,12 +1683,13 @@ public class ProcessService {
/**
* find process instance by the task id
+ *
* @param taskId taskId
* @return process instance
*/
- public ProcessInstance findProcessInstanceByTaskId(int taskId){
+ public ProcessInstance findProcessInstanceByTaskId(int taskId) {
TaskInstance taskInstance = taskInstanceMapper.selectById(taskId);
- if(taskInstance!= null){
+ if (taskInstance != null) {
return processInstanceMapper.selectById(taskInstance.getProcessInstanceId());
}
return null;
@@ -1639,97 +1697,101 @@ public class ProcessService {
/**
* find udf function list by id list string
+ *
* @param ids ids
* @return udf function list
*/
- public List<UdfFunc> queryUdfFunListByids(int[] ids){
+ public List<UdfFunc> queryUdfFunListByids(int[] ids) {
return udfFuncMapper.queryUdfByIdStr(ids, null);
}
/**
* find tenant code by resource name
- * @param resName resource name
+ *
+ * @param resName resource name
* @param resourceType resource type
* @return tenant code
*/
- public String queryTenantCodeByResName(String resName,ResourceType resourceType){
+ public String queryTenantCodeByResName(String resName, ResourceType resourceType) {
return resourceMapper.queryTenantCodeByResourceName(resName, resourceType.ordinal());
}
/**
* find schedule list by process define id.
+ *
* @param ids ids
* @return schedule list
*/
- public List<Schedule> selectAllByProcessDefineId(int[] ids){
+ public List<Schedule> selectAllByProcessDefineId(int[] ids) {
return scheduleMapper.selectAllByProcessDefineArray(
- ids);
+ ids);
}
/**
* get dependency cycle by work process define id and scheduler fire time
- * @param masterId masterId
+ *
+ * @param masterId masterId
* @param processDefinitionId processDefinitionId
- * @param scheduledFireTime the time the task schedule is expected to trigger
+ * @param scheduledFireTime the time the task schedule is expected to trigger
* @return CycleDependency
* @throws Exception if error throws Exception
*/
public CycleDependency getCycleDependency(int masterId, int processDefinitionId, Date scheduledFireTime) throws Exception {
- List<CycleDependency> list = getCycleDependencies(masterId,new int[]{processDefinitionId},scheduledFireTime);
- return list.size()>0 ? list.get(0) : null;
+ List<CycleDependency> list = getCycleDependencies(masterId, new int[] {processDefinitionId}, scheduledFireTime);
+ return list.size() > 0 ? list.get(0) : null;
}
/**
* get dependency cycle list by work process define id list and scheduler fire time
- * @param masterId masterId
- * @param ids ids
+ *
+ * @param masterId masterId
+ * @param ids ids
* @param scheduledFireTime the time the task schedule is expected to trigger
* @return CycleDependency list
* @throws Exception if error throws Exception
*/
- public List<CycleDependency> getCycleDependencies(int masterId,int[] ids,Date scheduledFireTime) throws Exception {
- List<CycleDependency> cycleDependencyList = new ArrayList<CycleDependency>();
+ public List<CycleDependency> getCycleDependencies(int masterId, int[] ids, Date scheduledFireTime) throws Exception {
+ List<CycleDependency> cycleDependencyList = new ArrayList<CycleDependency>();
if (ids == null || ids.length == 0) {
logger.warn("ids[] is empty!is invalid!");
return cycleDependencyList;
}
- if(scheduledFireTime == null){
+ if (scheduledFireTime == null) {
logger.warn("scheduledFireTime is null!is invalid!");
return cycleDependencyList;
}
-
String strCrontab = "";
CronExpression depCronExpression;
Cron depCron;
List<Date> list;
List<Schedule> schedules = this.selectAllByProcessDefineId(ids);
// for all scheduling information
- for(Schedule depSchedule:schedules){
+ for (Schedule depSchedule : schedules) {
strCrontab = depSchedule.getCrontab();
depCronExpression = CronUtils.parse2CronExpression(strCrontab);
depCron = CronUtils.parse2Cron(strCrontab);
CycleEnum cycleEnum = CronUtils.getMiniCycle(depCron);
- if(cycleEnum == null){
- logger.error("{} is not valid",strCrontab);
+ if (cycleEnum == null) {
+ logger.error("{} is not valid", strCrontab);
continue;
}
Calendar calendar = Calendar.getInstance();
- switch (cycleEnum){
+ switch (cycleEnum) {
/*case MINUTE:
calendar.add(Calendar.MINUTE,-61);*/
case HOUR:
- calendar.add(Calendar.HOUR,-25);
+ calendar.add(Calendar.HOUR, -25);
break;
case DAY:
- calendar.add(Calendar.DATE,-32);
+ calendar.add(Calendar.DATE, -32);
break;
case WEEK:
- calendar.add(Calendar.DATE,-32);
+ calendar.add(Calendar.DATE, -32);
break;
case MONTH:
- calendar.add(Calendar.MONTH,-13);
+ calendar.add(Calendar.MONTH, -13);
break;
default:
logger.warn("Dependent process definition's cycleEnum is {},not support!!", cycleEnum.name());
@@ -1737,14 +1799,14 @@ public class ProcessService {
}
Date start = calendar.getTime();
- if(depSchedule.getProcessDefinitionId() == masterId){
+ if (depSchedule.getProcessDefinitionId() == masterId) {
list = CronUtils.getSelfFireDateList(start, scheduledFireTime, depCronExpression);
- }else {
+ } else {
list = CronUtils.getFireDateList(start, scheduledFireTime, depCronExpression);
}
- if(list.size()>=1){
- start = list.get(list.size()-1);
- CycleDependency dependency = new CycleDependency(depSchedule.getProcessDefinitionId(),start, CronUtils.getExpirationTime(start, cycleEnum), cycleEnum);
+ if (list.size() >= 1) {
+ start = list.get(list.size() - 1);
+ CycleDependency dependency = new CycleDependency(depSchedule.getProcessDefinitionId(), start, CronUtils.getExpirationTime(start, cycleEnum), cycleEnum);
cycleDependencyList.add(dependency);
}
@@ -1754,78 +1816,80 @@ public class ProcessService {
/**
* find last scheduler process instance in the date interval
+ *
* @param definitionId definitionId
* @param dateInterval dateInterval
* @return process instance
*/
public ProcessInstance findLastSchedulerProcessInterval(int definitionId, DateInterval dateInterval) {
return processInstanceMapper.queryLastSchedulerProcess(definitionId,
- dateInterval.getStartTime(),
- dateInterval.getEndTime());
+ dateInterval.getStartTime(),
+ dateInterval.getEndTime());
}
/**
* find last manual process instance interval
+ *
* @param definitionId process definition id
* @param dateInterval dateInterval
* @return process instance
*/
public ProcessInstance findLastManualProcessInterval(int definitionId, DateInterval dateInterval) {
return processInstanceMapper.queryLastManualProcess(definitionId,
- dateInterval.getStartTime(),
- dateInterval.getEndTime());
+ dateInterval.getStartTime(),
+ dateInterval.getEndTime());
}
/**
* find last running process instance
- * @param definitionId process definition id
- * @param startTime start time
- * @param endTime end time
+ *
+ * @param definitionId process definition id
+ * @param startTime start time
+ * @param endTime end time
* @return process instance
*/
public ProcessInstance findLastRunningProcess(int definitionId, Date startTime, Date endTime) {
return processInstanceMapper.queryLastRunningProcess(definitionId,
- startTime,
- endTime,
- stateArray);
+ startTime,
+ endTime,
+ stateArray);
}
/**
* query user queue by process instance id
+ *
* @param processInstanceId processInstanceId
* @return queue
*/
- public String queryUserQueueByProcessInstanceId(int processInstanceId){
+ public String queryUserQueueByProcessInstanceId(int processInstanceId) {
String queue = "";
ProcessInstance processInstance = processInstanceMapper.selectById(processInstanceId);
- if(processInstance == null){
+ if (processInstance == null) {
return queue;
}
User executor = userMapper.selectById(processInstance.getExecutorId());
- if(executor != null){
+ if (executor != null) {
queue = executor.getQueue();
}
return queue;
}
-
-
/**
* get task worker group
+ *
* @param taskInstance taskInstance
* @return workerGroupId
*/
public String getTaskWorkerGroup(TaskInstance taskInstance) {
String workerGroup = taskInstance.getWorkerGroup();
-
- if(StringUtils.isNotBlank(workerGroup)){
+ if (StringUtils.isNotBlank(workerGroup)) {
return workerGroup;
}
int processInstanceId = taskInstance.getProcessInstanceId();
ProcessInstance processInstance = findProcessInstanceById(processInstanceId);
- if(processInstance != null){
+ if (processInstance != null) {
return processInstance.getWorkerGroup();
}
logger.info("task : {} will use default worker group", taskInstance.getId());
@@ -1834,18 +1898,19 @@ public class ProcessService {
/**
* get have perm project list
+ *
* @param userId userId
* @return project list
*/
- public List<Project> getProjectListHavePerm(int userId){
+ public List<Project> getProjectListHavePerm(int userId) {
List<Project> createProjects = projectMapper.queryProjectCreatedByUser(userId);
List<Project> authedProjects = projectMapper.queryAuthedProjectListByUserId(userId);
- if(createProjects == null){
+ if (createProjects == null) {
createProjects = new ArrayList<>();
}
- if(authedProjects != null){
+ if (authedProjects != null) {
createProjects.addAll(authedProjects);
}
return createProjects;
@@ -1853,13 +1918,14 @@ public class ProcessService {
/**
* get have perm project ids
+ *
* @param userId userId
* @return project ids
*/
- public List<Integer> getProjectIdListHavePerm(int userId){
+ public List<Integer> getProjectIdListHavePerm(int userId) {
List<Integer> projectIdList = new ArrayList<>();
- for(Project project : getProjectListHavePerm(userId)){
+ for (Project project : getProjectListHavePerm(userId)) {
projectIdList.add(project.getId());
}
return projectIdList;
@@ -1867,17 +1933,18 @@ public class ProcessService {
/**
* list unauthorized udf function
- * @param userId user id
- * @param needChecks data source id array
+ *
+ * @param userId user id
+ * @param needChecks data source id array
* @return unauthorized udf function list
*/
- public <T> List<T> listUnauthorized(int userId,T[] needChecks,AuthorizationType authorizationType){
+ public <T> List<T> listUnauthorized(int userId, T[] needChecks, AuthorizationType authorizationType) {
List<T> resultList = new ArrayList<T>();
if (Objects.nonNull(needChecks) && needChecks.length > 0) {
Set<T> originResSet = new HashSet<T>(Arrays.asList(needChecks));
- switch (authorizationType){
+ switch (authorizationType) {
case RESOURCE_FILE_ID:
Set<Integer> authorizedResourceFiles = resourceMapper.listAuthorizedResourceById(userId, needChecks).stream().map(t -> t.getId()).collect(toSet());
originResSet.removeAll(authorizedResourceFiles);
@@ -1891,13 +1958,15 @@ public class ProcessService {
originResSet.removeAll(authorizedUdfFiles);
break;
case DATASOURCE:
- Set<Integer> authorizedDatasources = dataSourceMapper.listAuthorizedDataSource(userId,needChecks).stream().map(t -> t.getId()).collect(toSet());
+ Set<Integer> authorizedDatasources = dataSourceMapper.listAuthorizedDataSource(userId, needChecks).stream().map(t -> t.getId()).collect(toSet());
originResSet.removeAll(authorizedDatasources);
break;
case UDF:
Set<Integer> authorizedUdfs = udfFuncMapper.listAuthorizedUdfFunc(userId, needChecks).stream().map(t -> t.getId()).collect(toSet());
originResSet.removeAll(authorizedUdfs);
break;
+ default:
+ break;
}
resultList.addAll(originResSet);
@@ -1908,48 +1977,51 @@ public class ProcessService {
/**
* get user by user id
+ *
* @param userId user id
* @return User
*/
- public User getUserById(int userId){
+ public User getUserById(int userId) {
return userMapper.selectById(userId);
}
/**
* get resource by resoruce id
+ *
* @param resoruceId resource id
* @return Resource
*/
- public Resource getResourceById(int resoruceId){
+ public Resource getResourceById(int resoruceId) {
return resourceMapper.selectById(resoruceId);
}
-
/**
* list resources by ids
+ *
* @param resIds resIds
* @return resource list
*/
- public List<Resource> listResourceByIds(Integer[] resIds){
+ public List<Resource> listResourceByIds(Integer[] resIds) {
return resourceMapper.listResourceByIds(resIds);
}
/**
* format task app id in task instance
+ *
* @param taskInstance
* @return
*/
- public String formatTaskAppId(TaskInstance taskInstance){
+ public String formatTaskAppId(TaskInstance taskInstance) {
ProcessDefinition definition = this.findProcessDefineById(taskInstance.getProcessDefinitionId());
ProcessInstance processInstanceById = this.findProcessInstanceById(taskInstance.getProcessInstanceId());
- if(definition == null || processInstanceById == null){
+ if (definition == null || processInstanceById == null) {
return "";
}
return String.format("%s_%s_%s",
- definition.getId(),
- processInstanceById.getId(),
- taskInstance.getId());
+ definition.getId(),
+ processInstanceById.getId(),
+ taskInstance.getId());
}
}
diff --git a/pom.xml b/pom.xml
index d37492e..cd281ba 100644
--- a/pom.xml
+++ b/pom.xml
@@ -940,7 +940,6 @@
<include>**/dao/mapper/TenantMapperTest.java</include>
<include>**/dao/mapper/UdfFuncMapperTest.java</include>
<include>**/dao/mapper/UDFUserMapperTest.java</include>
- <include>**/dao/mapper/UserAlertGroupMapperTest.java</include>
<include>**/dao/mapper/UserMapperTest.java</include>
<include>**/dao/mapper/AlertPluginInstanceMapperTest.java</include>
<include>**/dao/mapper/PluginDefineTest.java</include>
diff --git a/sql/dolphinscheduler-postgre.sql b/sql/dolphinscheduler-postgre.sql
index 2149b6c..078ff7e 100644
--- a/sql/dolphinscheduler-postgre.sql
+++ b/sql/dolphinscheduler-postgre.sql
@@ -187,14 +187,10 @@ DROP TABLE IF EXISTS t_ds_alert;
CREATE TABLE t_ds_alert (
id int NOT NULL ,
title varchar(64) DEFAULT NULL ,
- show_type int DEFAULT NULL ,
content text ,
- alert_type int DEFAULT NULL ,
alert_status int DEFAULT '0' ,
log text ,
alertgroup_id int DEFAULT NULL ,
- receivers text ,
- receivers_cc text ,
create_time timestamp DEFAULT NULL ,
update_time timestamp DEFAULT NULL ,
PRIMARY KEY (id)
@@ -204,14 +200,14 @@ CREATE TABLE t_ds_alert (
--
DROP TABLE IF EXISTS t_ds_alertgroup;
-CREATE TABLE t_ds_alertgroup (
- id int NOT NULL ,
- group_name varchar(255) DEFAULT NULL ,
- group_type int DEFAULT NULL ,
- description varchar(255) DEFAULT NULL ,
- create_time timestamp DEFAULT NULL ,
- update_time timestamp DEFAULT NULL ,
- PRIMARY KEY (id)
+CREATE TABLE t_ds_alertgroup(
+ id int NOT NULL,
+ create_user_id int4 DEFAULT NULL,
+ group_name varchar(255) DEFAULT NULL,
+ description varchar(255) DEFAULT NULL,
+ create_time timestamp DEFAULT NULL,
+ update_time timestamp DEFAULT NULL,
+ PRIMARY KEY (id)
) ;
--
@@ -301,8 +297,7 @@ CREATE TABLE t_ds_process_definition (
flag int DEFAULT NULL ,
locations text ,
connects text ,
- receivers text ,
- receivers_cc text ,
+ warning_group_id int4 DEFAULT NULL ,
create_time timestamp DEFAULT NULL ,
timeout int DEFAULT '0' ,
tenant_id int NOT NULL DEFAULT '-1' ,
@@ -329,8 +324,7 @@ CREATE TABLE t_ds_process_definition_version (
global_params text ,
locations text ,
connects text ,
- receivers text ,
- receivers_cc text ,
+ warning_group_id int4 DEFAULT NULL,
create_time timestamp DEFAULT NULL ,
timeout int DEFAULT '0' ,
resource_ids varchar(64),
@@ -493,20 +487,6 @@ CREATE TABLE t_ds_relation_udfs_user (
;
--
--- Table structure for table t_ds_relation_user_alertgroup
---
-
-DROP TABLE IF EXISTS t_ds_relation_user_alertgroup;
-CREATE TABLE t_ds_relation_user_alertgroup (
- id int NOT NULL,
- alertgroup_id int DEFAULT NULL,
- user_id int DEFAULT NULL,
- create_time timestamp DEFAULT NULL,
- update_time timestamp DEFAULT NULL,
- PRIMARY KEY (id)
-);
-
---
-- Table structure for table t_ds_resources
--
@@ -748,9 +728,6 @@ ALTER TABLE t_ds_relation_resources_user ALTER COLUMN id SET DEFAULT NEXTVAL('t_
DROP SEQUENCE IF EXISTS t_ds_relation_udfs_user_id_sequence;
CREATE SEQUENCE t_ds_relation_udfs_user_id_sequence;
ALTER TABLE t_ds_relation_udfs_user ALTER COLUMN id SET DEFAULT NEXTVAL('t_ds_relation_udfs_user_id_sequence');
-DROP SEQUENCE IF EXISTS t_ds_relation_user_alertgroup_id_sequence;
-CREATE SEQUENCE t_ds_relation_user_alertgroup_id_sequence;
-ALTER TABLE t_ds_relation_user_alertgroup ALTER COLUMN id SET DEFAULT NEXTVAL('t_ds_relation_user_alertgroup_id_sequence');
DROP SEQUENCE IF EXISTS t_ds_resources_id_sequence;
CREATE SEQUENCE t_ds_resources_id_sequence;
@@ -777,21 +754,27 @@ ALTER TABLE t_ds_version ALTER COLUMN id SET DEFAULT NEXTVAL('t_ds_version_id_se
DROP SEQUENCE IF EXISTS t_ds_worker_group_id_sequence;
CREATE SEQUENCE t_ds_worker_group_id_sequence;
-ALTER TABLE t_ds_worker_group ALTER COLUMN id SET DEFAULT NEXTVAL('t_ds_worker_group_id_sequence');
+ALTER TABLE t_ds_worker_group
+ ALTER COLUMN id SET DEFAULT NEXTVAL('t_ds_worker_group_id_sequence');
DROP SEQUENCE IF EXISTS t_ds_worker_server_id_sequence;
-CREATE SEQUENCE t_ds_worker_server_id_sequence;
-ALTER TABLE t_ds_worker_server ALTER COLUMN id SET DEFAULT NEXTVAL('t_ds_worker_server_id_sequence');
+CREATE SEQUENCE t_ds_worker_server_id_sequence;
+ALTER TABLE t_ds_worker_server
+ ALTER COLUMN id SET DEFAULT NEXTVAL('t_ds_worker_server_id_sequence');
-- Records of t_ds_user?user : admin , password : dolphinscheduler123
-INSERT INTO t_ds_user(user_name,user_password,user_type,email,phone,tenant_id,state,create_time,update_time) VALUES ('admin', '7ad2410b2f4c074479a8937a28a22b8f', '0', 'xxx@qq.com', '', '0', 1, '2018-03-27 15:48:50', '2018-10-24 17:40:22');
+INSERT INTO t_ds_user(user_name, user_password, user_type, email, phone, tenant_id, state, create_time, update_time)
+VALUES ('admin', '7ad2410b2f4c074479a8937a28a22b8f', '0', 'xxx@qq.com', '', '0', 1, '2018-03-27 15:48:50',
+ '2018-10-24 17:40:22');
-- Records of t_ds_alertgroup，dolphinscheduler warning group
-INSERT INTO t_ds_alertgroup(group_name,group_type,description,create_time,update_time) VALUES ('dolphinscheduler warning group', '0', 'dolphinscheduler warning group','2018-11-29 10:20:39', '2018-11-29 10:20:39');
-INSERT INTO t_ds_relation_user_alertgroup(alertgroup_id,user_id,create_time,update_time) VALUES ( '1', '1', '2018-11-29 10:22:33', '2018-11-29 10:22:33');
+INSERT INTO t_ds_alertgroup(id, create_user_id, group_name, description, create_time, update_time)
+VALUES (1, 'dolphinscheduler warning group', 'dolphinscheduler warning group', '2018-11-29 10:20:39',
+ '2018-11-29 10:20:39');
-- Records of t_ds_queue,default queue name : default
-INSERT INTO t_ds_queue(queue_name,queue,create_time,update_time) VALUES ('default', 'default','2018-11-29 10:22:33', '2018-11-29 10:22:33');
+INSERT INTO t_ds_queue(queue_name, queue, create_time, update_time)
+VALUES ('default', 'default', '2018-11-29 10:22:33', '2018-11-29 10:22:33');
-- Records of t_ds_queue,default queue name : default
INSERT INTO t_ds_version(version) VALUES ('1.3.0');
diff --git a/sql/dolphinscheduler_mysql.sql b/sql/dolphinscheduler_mysql.sql
index 7abdd39..986df1f 100644
--- a/sql/dolphinscheduler_mysql.sql
+++ b/sql/dolphinscheduler_mysql.sql
@@ -279,14 +279,10 @@ DROP TABLE IF EXISTS `t_ds_alert`;
CREATE TABLE `t_ds_alert` (
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'key',
`title` varchar(64) DEFAULT NULL COMMENT 'title',
- `show_type` tinyint(4) DEFAULT NULL COMMENT 'send email type,0:TABLE,1:TEXT',
`content` text COMMENT 'Message content (can be email, can be SMS. Mail is stored in JSON map, and SMS is string)',
- `alert_type` tinyint(4) DEFAULT NULL COMMENT '0:email,1:sms',
`alert_status` tinyint(4) DEFAULT '0' COMMENT '0:wait running,1:success,2:failed',
`log` text COMMENT 'log',
`alertgroup_id` int(11) DEFAULT NULL COMMENT 'alert group id',
- `receivers` text COMMENT 'receivers',
- `receivers_cc` text COMMENT 'cc',
`create_time` datetime DEFAULT NULL COMMENT 'create time',
`update_time` datetime DEFAULT NULL COMMENT 'update time',
PRIMARY KEY (`id`)
@@ -300,14 +296,14 @@ CREATE TABLE `t_ds_alert` (
-- Table structure for t_ds_alertgroup
-- ----------------------------
DROP TABLE IF EXISTS `t_ds_alertgroup`;
-CREATE TABLE `t_ds_alertgroup` (
- `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'key',
- `group_name` varchar(255) DEFAULT NULL COMMENT 'group name',
- `group_type` tinyint(4) DEFAULT NULL COMMENT 'Group type (message 0, SMS 1...)',
- `description` varchar(255) DEFAULT NULL,
- `create_time` datetime DEFAULT NULL COMMENT 'create time',
- `update_time` datetime DEFAULT NULL COMMENT 'update time',
- PRIMARY KEY (`id`)
+CREATE TABLE `t_ds_alertgroup`(
+ `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'key',
+ `create_user_id` int(11) DEFAULT NULL COMMENT 'create user id',
+ `group_name` varchar(255) DEFAULT NULL COMMENT 'group name',
+ `description` varchar(255) DEFAULT NULL,
+ `create_time` datetime DEFAULT NULL COMMENT 'create time',
+ `update_time` datetime DEFAULT NULL COMMENT 'update time',
+ PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
-- ----------------------------
@@ -406,8 +402,7 @@ CREATE TABLE `t_ds_process_definition` (
`flag` tinyint(4) DEFAULT NULL COMMENT '0 not available, 1 available',
`locations` text COMMENT 'Node location information',
`connects` text COMMENT 'Node connection information',
- `receivers` text COMMENT 'receivers',
- `receivers_cc` text COMMENT 'cc',
+ `warning_group_id` int(11) DEFAULT NULL COMMENT 'alert group id',
`create_time` datetime DEFAULT NULL COMMENT 'create time',
`timeout` int(11) DEFAULT '0' COMMENT 'time out',
`tenant_id` int(11) NOT NULL DEFAULT '-1' COMMENT 'tenant id',
@@ -436,8 +431,7 @@ CREATE TABLE `t_ds_process_definition_version` (
`global_params` text COMMENT 'global parameters',
`locations` text COMMENT 'Node location information',
`connects` text COMMENT 'Node connection information',
- `receivers` text COMMENT 'receivers',
- `receivers_cc` text COMMENT 'cc',
+ `warning_group_id` int(11) DEFAULT NULL COMMENT 'alert group id',
`create_time` datetime DEFAULT NULL COMMENT 'create time',
`timeout` int(11) DEFAULT '0' COMMENT 'time out',
`resource_ids` varchar(255) DEFAULT NULL COMMENT 'resource ids',
@@ -621,27 +615,6 @@ CREATE TABLE `t_ds_relation_udfs_user` (
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
-- ----------------------------
--- Records of t_ds_relation_udfs_user
--- ----------------------------
-
--- ----------------------------
--- Table structure for t_ds_relation_user_alertgroup
--- ----------------------------
-DROP TABLE IF EXISTS `t_ds_relation_user_alertgroup`;
-CREATE TABLE `t_ds_relation_user_alertgroup` (
- `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'key',
- `alertgroup_id` int(11) DEFAULT NULL COMMENT 'alert group id',
- `user_id` int(11) DEFAULT NULL COMMENT 'user id',
- `create_time` datetime DEFAULT NULL COMMENT 'create time',
- `update_time` datetime DEFAULT NULL COMMENT 'update time',
- PRIMARY KEY (`id`)
-) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
-
--- ----------------------------
--- Records of t_ds_relation_user_alertgroup
--- ----------------------------
-
--- ----------------------------
-- Table structure for t_ds_resources
-- ----------------------------
DROP TABLE IF EXISTS `t_ds_resources`;
@@ -831,28 +804,29 @@ CREATE TABLE `t_ds_version` (
-- ----------------------------
-- Records of t_ds_version
-- ----------------------------
-INSERT INTO `t_ds_version` VALUES ('1', '1.3.0');
+INSERT INTO `t_ds_version`
+VALUES ('1', '1.3.0');
-- ----------------------------
-- Records of t_ds_alertgroup
-- ----------------------------
-INSERT INTO `t_ds_alertgroup` VALUES ('1', 'default admin warning group', '0', 'default admin warning group', '2018-11-29 10:20:39', '2018-11-29 10:20:39');
-
--- ----------------------------
--- Records of t_ds_relation_user_alertgroup
--- ----------------------------
-INSERT INTO `t_ds_relation_user_alertgroup` VALUES ('1', '1', '1', '2018-11-29 10:22:33', '2018-11-29 10:22:33');
+INSERT INTO `t_ds_alertgroup`
+VALUES (1, 1, 'default admin warning group', 'default admin warning group', '2018-11-29 10:20:39',
+ '2018-11-29 10:20:39');
-- ----------------------------
-- Records of t_ds_user
-- ----------------------------
-INSERT INTO `t_ds_user` VALUES ('1', 'admin', '7ad2410b2f4c074479a8937a28a22b8f', '0', 'xxx@qq.com', '', '0', '2018-03-27 15:48:50', '2018-10-24 17:40:22', null, 1);
+INSERT INTO `t_ds_user`
+VALUES ('1', 'admin', '7ad2410b2f4c074479a8937a28a22b8f', '0', 'xxx@qq.com', '', '0', '2018-03-27 15:48:50',
+ '2018-10-24 17:40:22', null, 1);
-- ----------------------------
-- Table structure for t_ds_plugin_define
-- ----------------------------
-SET sql_mode=(SELECT REPLACE(@@sql_mode,'ONLY_FULL_GROUP_BY',''));
+SET
+sql_mode=(SELECT REPLACE(@@sql_mode,'ONLY_FULL_GROUP_BY',''));
DROP TABLE IF EXISTS `t_ds_plugin_define`;
CREATE TABLE `t_ds_plugin_define` (
`id` int NOT NULL AUTO_INCREMENT,
diff --git a/sql/upgrade/1.4.0_schema/mysql/dolphinscheduler_ddl.sql b/sql/upgrade/1.4.0_schema/mysql/dolphinscheduler_ddl.sql
index 42e62fb..a0315dd 100644
--- a/sql/upgrade/1.4.0_schema/mysql/dolphinscheduler_ddl.sql
+++ b/sql/upgrade/1.4.0_schema/mysql/dolphinscheduler_ddl.sql
@@ -30,14 +30,40 @@ CREATE TABLE `t_ds_plugin_define` (
DROP TABLE IF EXISTS `t_ds_alert_plugin_instance`;
CREATE TABLE `t_ds_alert_plugin_instance` (
- `id` int NOT NULL AUTO_INCREMENT,
- `plugin_define_id` int NOT NULL,
- `plugin_instance_params` text COMMENT 'plugin instance params. Also contain the params value which user input in web ui.',
- `create_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP,
- `update_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
- `alert_group_id` int DEFAULT NULL,
- `instance_name` varchar(200) DEFAULT NULL COMMENT 'alert instance name',
- PRIMARY KEY (`id`)
+ `id` int NOT NULL AUTO_INCREMENT,
+ `plugin_define_id` int NOT NULL,
+ `plugin_instance_params` text COMMENT 'plugin instance params. Also contain the params value which user input in web ui.',
+ `create_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP,
+ `update_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+ `alert_group_id` int DEFAULT NULL,
+ `instance_name` varchar(200) DEFAULT NULL COMMENT 'alert instance name',
+ PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
+ALTER TABLE t_ds_process_definition
+ ADD COLUMN `warning_group_id` int(11) DEFAULT NULL COMMENT 'alert group id' AFTER `connects`;
+
+ALTER TABLE t_ds_process_definition_version
+ ADD COLUMN `warning_group_id` int(11) DEFAULT NULL COMMENT 'alert group id' AFTER `connects`;
+
+ALTER TABLE t_ds_alertgroup
+ ADD COLUMN `create_user_id` int(11) DEFAULT NULL COMMENT 'create user id' AFTER `id`;
+
+
+-- ----------------------------
+-- These columns will not be used in the new version,if you determine that the historical data is useless, you can delete it using the sql below
+-- ----------------------------
+/*
+ALTER TABLE t_ds_process_definition DROP receivers, DROP receivers_cc;
+
+ALTER TABLE t_ds_process_definition_version DROP receivers, DROP receivers_cc;
+
+ALTER TABLE t_ds_alert DROP show_type,DROP alert_type,DROP receivers,DROP receivers_cc;
+
+ALTER TABLE t_ds_alertgroup DROP group_type;
+
+DROP TABLE IF EXISTS t_ds_relation_user_alertgroup;
+*/
+
+
diff --git a/sql/upgrade/1.4.0_schema/postgresql/dolphinscheduler_ddl.sql b/sql/upgrade/1.4.0_schema/postgresql/dolphinscheduler_ddl.sql
index 90de08f..788347b 100644
--- a/sql/upgrade/1.4.0_schema/postgresql/dolphinscheduler_ddl.sql
+++ b/sql/upgrade/1.4.0_schema/postgresql/dolphinscheduler_ddl.sql
@@ -29,12 +29,36 @@ CREATE TABLE t_ds_plugin_define (
DROP TABLE IF EXISTS t_ds_alert_plugin_instance;
CREATE TABLE t_ds_alert_plugin_instance (
- id serial NOT NULL,
- plugin_define_id int4 NOT NULL,
- plugin_instance_params text NULL,
- create_time timestamp NULL,
- update_time timestamp NULL,
- alert_group_id int4 NOT NULL,
- instance_name varchar(200) NULL,
- CONSTRAINT t_ds_alert_plugin_instance_pk PRIMARY KEY (id)
+ id serial NOT NULL,
+ plugin_define_id int4 NOT NULL,
+ plugin_instance_params text NULL,
+ create_time timestamp NULL,
+ update_time timestamp NULL,
+ alert_group_id int4 NOT NULL,
+ instance_name varchar(200) NULL,
+ CONSTRAINT t_ds_alert_plugin_instance_pk PRIMARY KEY (id)
);
+
+ALTER TABLE t_ds_process_definition
+ ADD COLUMN `warning_group_id` int4 DEFAULT NULL COMMENT 'alert group id' AFTER `connects`;
+
+ALTER TABLE t_ds_process_definition_version
+ ADD COLUMN `warning_group_id` int4 DEFAULT NULL COMMENT 'alert group id' AFTER `connects`;
+
+ALTER TABLE t_ds_alertgroup
+ ADD COLUMN `create_user_id` int4 DEFAULT NULL COMMENT 'create user id' AFTER `id`;
+
+-- ----------------------------
+-- These columns will not be used in the new version,if you determine that the historical data is useless, you can delete it using the sql below
+-- ----------------------------
+/*
+ALTER TABLE t_ds_process_definition DROP COLUMN "receivers", DROP COLUMN "receivers_cc";
+
+ALTER TABLE t_ds_process_definition_version DROP COLUMN "receivers", DROP COLUMN "receivers_cc";
+
+ALTER TABLE t_ds_alert DROP COLUMN "show_type",DROP COLUMN "alert_type",DROP COLUMN "receivers",DROP COLUMN "receivers_cc";
+
+ALTER TABLE t_ds_alertgroup DROP COLUMN "group_type";
+
+DROP TABLE IF EXISTS t_ds_relation_user_alertgroup;
+*/