You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by we...@apache.org on 2024/03/21 09:07:09 UTC
(dolphinscheduler) branch dev updated: [Bug] Disable delete a specific workflow version when exist workflow instance under this version which is not finish (#15730)
This is an automated email from the ASF dual-hosted git repository.
wenjun pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new 56a834b48b [Bug] Disable delete a specific workflow version when exist workflow instance under this version which is not finish (#15730)
56a834b48b is described below
commit 56a834b48bb6b2225dff28c5f69a7bac0326c146
Author: Wenjun Ruan <we...@apache.org>
AuthorDate: Thu Mar 21 17:07:03 2024 +0800
[Bug] Disable delete a specific workflow version when exist workflow instance under this version which is not finish (#15730)
---
.../controller/ProcessDefinitionController.java | 148 +++++++--------
.../api/service/ProcessDefinitionService.java | 151 ++++++++--------
.../api/service/ProcessInstanceService.java | 76 ++++----
.../service/impl/ProcessDefinitionServiceImpl.java | 200 ++++++++++-----------
.../service/impl/ProcessInstanceServiceImpl.java | 7 +
.../ProcessDefinitionControllerTest.java | 18 +-
.../common/enums/WorkflowExecutionStatus.java | 15 ++
.../dao/mapper/ProcessInstanceMapper.java | 4 +
.../dao/repository/ProcessInstanceDao.java | 10 +-
.../repository/impl/ProcessInstanceDaoImpl.java | 12 +-
.../dao/mapper/ProcessInstanceMapper.xml | 15 ++
.../impl/ProcessInstanceDaoImplTest.java | 104 +++++++++++
12 files changed, 463 insertions(+), 297 deletions(-)
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ProcessDefinitionController.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ProcessDefinitionController.java
index a7701d25b3..261ea9183b 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ProcessDefinitionController.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ProcessDefinitionController.java
@@ -90,16 +90,16 @@ public class ProcessDefinitionController extends BaseController {
/**
* create process definition
*
- * @param loginUser login user
- * @param projectCode project code
- * @param name process definition name
- * @param description description
- * @param globalParams globalParams
- * @param locations locations for nodes
- * @param timeout timeout
- * @param taskRelationJson relation json for nodes
+ * @param loginUser login user
+ * @param projectCode project code
+ * @param name process definition name
+ * @param description description
+ * @param globalParams globalParams
+ * @param locations locations for nodes
+ * @param timeout timeout
+ * @param taskRelationJson relation json for nodes
* @param taskDefinitionJson taskDefinitionJson
- * @param otherParamsJson otherParamsJson handle other params
+ * @param otherParamsJson otherParamsJson handle other params
* @return create result code
*/
@Operation(summary = "createProcessDefinition", description = "CREATE_PROCESS_DEFINITION_NOTES")
@@ -132,9 +132,9 @@ public class ProcessDefinitionController extends BaseController {
/**
* copy process definition
*
- * @param loginUser login user
- * @param projectCode project code
- * @param codes process definition codes
+ * @param loginUser login user
+ * @param projectCode project code
+ * @param codes process definition codes
* @param targetProjectCode target project code
* @return copy result code
*/
@@ -157,9 +157,9 @@ public class ProcessDefinitionController extends BaseController {
/**
* move process definition
*
- * @param loginUser login user
- * @param projectCode project code
- * @param codes process definition codes
+ * @param loginUser login user
+ * @param projectCode project code
+ * @param codes process definition codes
* @param targetProjectCode target project code
* @return move result code
*/
@@ -182,9 +182,9 @@ public class ProcessDefinitionController extends BaseController {
/**
* verify process definition name unique
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param name name
+ * @param name name
* @return true if process definition name not exists, otherwise false
*/
@Operation(summary = "verify-name", description = "VERIFY_PROCESS_DEFINITION_NAME_NOTES")
@@ -207,17 +207,17 @@ public class ProcessDefinitionController extends BaseController {
/**
* update process definition, with whole process definition object including task definition, task relation and location.
*
- * @param loginUser login user
- * @param projectCode project code
- * @param name process definition name
- * @param code process definition code
- * @param description description
- * @param globalParams globalParams
- * @param locations locations for nodes
- * @param timeout timeout
- * @param taskRelationJson relation json for nodes
+ * @param loginUser login user
+ * @param projectCode project code
+ * @param name process definition name
+ * @param code process definition code
+ * @param description description
+ * @param globalParams globalParams
+ * @param locations locations for nodes
+ * @param timeout timeout
+ * @param taskRelationJson relation json for nodes
* @param taskDefinitionJson taskDefinitionJson
- * @param otherParamsJson otherParamsJson handle other params
+ * @param otherParamsJson otherParamsJson handle other params
* @return update result code
*/
@Operation(summary = "update", description = "UPDATE_PROCESS_DEFINITION_NOTES")
@@ -263,11 +263,11 @@ public class ProcessDefinitionController extends BaseController {
/**
* query process definition version paging list info
*
- * @param loginUser login user info
+ * @param loginUser login user info
* @param projectCode project code
- * @param pageNo the process definition version list current page number
- * @param pageSize the process definition version list page size
- * @param code the process definition code
+ * @param pageNo the process definition version list current page number
+ * @param pageSize the process definition version list page size
+ * @param code the process definition code
* @return the process definition version list
*/
@Operation(summary = "queryVersions", description = "QUERY_PROCESS_DEFINITION_VERSIONS_NOTES")
@@ -293,10 +293,10 @@ public class ProcessDefinitionController extends BaseController {
/**
* switch certain process definition version
*
- * @param loginUser login user info
+ * @param loginUser login user info
* @param projectCode project code
- * @param code the process definition code
- * @param version the version user want to switch
+ * @param code the process definition code
+ * @param version the version user want to switch
* @return switch version result code
*/
@Operation(summary = "switchVersion", description = "SWITCH_PROCESS_DEFINITION_VERSION_NOTES")
@@ -319,10 +319,10 @@ public class ProcessDefinitionController extends BaseController {
/**
* delete the certain process definition version by version and process definition code
*
- * @param loginUser login user info
+ * @param loginUser login user info
* @param projectCode project code
- * @param code the process definition code
- * @param version the process definition version user want to delete
+ * @param code the process definition code
+ * @param version the process definition version user want to delete
* @return delete version result code
*/
@Operation(summary = "deleteVersion", description = "DELETE_PROCESS_DEFINITION_VERSION_NOTES")
@@ -333,13 +333,13 @@ public class ProcessDefinitionController extends BaseController {
@DeleteMapping(value = "/{code}/versions/{version}")
@ResponseStatus(HttpStatus.OK)
@ApiException(DELETE_PROCESS_DEFINITION_VERSION_ERROR)
- public Result deleteProcessDefinitionVersion(@Parameter(hidden = true) @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
- @Parameter(name = "projectCode", description = "PROJECT_CODE", required = true) @PathVariable long projectCode,
- @PathVariable(value = "code") long code,
- @PathVariable(value = "version") int version) {
- Map<String, Object> result =
- processDefinitionService.deleteProcessDefinitionVersion(loginUser, projectCode, code, version);
- return returnDataList(result);
+ public Result<Void> deleteProcessDefinitionVersion(@Parameter(hidden = true) @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
+ @Parameter(name = "projectCode", description = "PROJECT_CODE", required = true) @PathVariable long projectCode,
+ @PathVariable(value = "code") long workflowDefinitionCode,
+ @PathVariable(value = "version") int workflowDefinitionVersion) {
+ processDefinitionService.deleteProcessDefinitionVersion(loginUser, projectCode, workflowDefinitionCode,
+ workflowDefinitionVersion);
+ return Result.success();
}
@Operation(summary = "release", description = "RELEASE_PROCESS_DEFINITION_NOTES")
@@ -372,9 +372,9 @@ public class ProcessDefinitionController extends BaseController {
/**
* query detail of process definition by code
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param code process definition code
+ * @param code process definition code
* @return process definition detail
*/
@Operation(summary = "queryProcessDefinitionByCode", description = "QUERY_PROCESS_DEFINITION_BY_CODE_NOTES")
@@ -395,9 +395,9 @@ public class ProcessDefinitionController extends BaseController {
/**
* query detail of process definition by name
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param name process definition name
+ * @param name process definition name
* @return process definition detail
*/
@Operation(summary = "queryProcessDefinitionByName", description = "QUERY_PROCESS_DEFINITION_BY_NAME_NOTES")
@@ -418,7 +418,7 @@ public class ProcessDefinitionController extends BaseController {
/**
* query Process definition list
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
* @return process definition list
*/
@@ -435,7 +435,7 @@ public class ProcessDefinitionController extends BaseController {
/**
* query Process definition simple list
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
* @return process definition list
*/
@@ -452,13 +452,13 @@ public class ProcessDefinitionController extends BaseController {
/**
* query process definition list paging
*
- * @param loginUser login user
- * @param projectCode project code
- * @param searchVal search value
+ * @param loginUser login user
+ * @param projectCode project code
+ * @param searchVal search value
* @param otherParamsJson otherParamsJson handle other params
- * @param pageNo page number
- * @param pageSize page size
- * @param userId user id
+ * @param pageNo page number
+ * @param pageSize page size
+ * @param userId user id
* @return process definition page
*/
@Operation(summary = "queryListPaging", description = "QUERY_PROCESS_DEFINITION_LIST_PAGING_NOTES")
@@ -493,10 +493,10 @@ public class ProcessDefinitionController extends BaseController {
/**
* encapsulation tree view structure
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param code process definition code
- * @param limit limit
+ * @param code process definition code
+ * @param limit limit
* @return tree view json data
*/
@Operation(summary = "viewTree", description = "VIEW_TREE_NOTES")
@@ -518,9 +518,9 @@ public class ProcessDefinitionController extends BaseController {
/**
* get tasks list by process definition code
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param code process definition code
+ * @param code process definition code
* @return task list
*/
@Operation(summary = "getTasksByDefinitionCode", description = "GET_TASK_LIST_BY_DEFINITION_CODE_NOTES")
@@ -541,9 +541,9 @@ public class ProcessDefinitionController extends BaseController {
/**
* get tasks list map by process definition multiple code
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param codes process definition codes
+ * @param codes process definition codes
* @return node list data
*/
@Operation(summary = "getTaskListByDefinitionCodes", description = "GET_TASK_LIST_BY_DEFINITION_CODE_NOTES")
@@ -564,7 +564,7 @@ public class ProcessDefinitionController extends BaseController {
/**
* get process definition list map by project code
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
* @return process definition list data
*/
@@ -584,7 +584,7 @@ public class ProcessDefinitionController extends BaseController {
/**
* get task definition list by process definition code
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
* @return process definition list data
*/
@@ -621,9 +621,9 @@ public class ProcessDefinitionController extends BaseController {
/**
* batch delete process definition by codes
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param codes process definition code list
+ * @param codes process definition code list
* @return delete result code
*/
@Operation(summary = "batchDeleteByCodes", description = "BATCH_DELETE_PROCESS_DEFINITION_BY_IDS_NOTES")
@@ -645,10 +645,10 @@ public class ProcessDefinitionController extends BaseController {
/**
* batch export process definition by codes
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param codes process definition codes
- * @param response response
+ * @param codes process definition codes
+ * @param response response
*/
@Operation(summary = "batchExportByCodes", description = "BATCH_EXPORT_PROCESS_DEFINITION_BY_CODES_NOTES")
@Parameters({
@@ -670,7 +670,7 @@ public class ProcessDefinitionController extends BaseController {
/**
* query all process definition by project code
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
* @return process definition list
*/
@@ -688,9 +688,9 @@ public class ProcessDefinitionController extends BaseController {
/**
* import process definition
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param file resource file
+ * @param file resource file
* @return import result code
*/
@Operation(summary = "importProcessDefinition", description = "IMPORT_PROCESS_DEFINITION_NOTES")
@@ -715,7 +715,7 @@ public class ProcessDefinitionController extends BaseController {
* query process definition global variables and local variables
*
* @param loginUser login user
- * @param code process definition code
+ * @param code process definition code
* @return variables data
*/
@Operation(summary = "viewVariables", description = "QUERY_PROCESS_DEFINITION_GLOBAL_VARIABLES_AND_LOCAL_VARIABLES_NOTES")
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionService.java
index cf28e465ef..52a95839ba 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionService.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionService.java
@@ -43,16 +43,16 @@ public interface ProcessDefinitionService {
/**
* create process definition
*
- * @param loginUser login user
- * @param projectCode project code
- * @param name process definition name
- * @param description description
- * @param globalParams global params
- * @param locations locations for nodes
- * @param timeout timeout
- * @param taskRelationJson relation json for nodes
+ * @param loginUser login user
+ * @param projectCode project code
+ * @param name process definition name
+ * @param description description
+ * @param globalParams global params
+ * @param locations locations for nodes
+ * @param timeout timeout
+ * @param taskRelationJson relation json for nodes
* @param taskDefinitionJson taskDefinitionJson
- * @param otherParamsJson otherParamsJson handle other params
+ * @param otherParamsJson otherParamsJson handle other params
* @return create result code
*/
Map<String, Object> createProcessDefinition(User loginUser,
@@ -70,7 +70,7 @@ public interface ProcessDefinitionService {
/**
* create process definition V2
*
- * @param loginUser login user
+ * @param loginUser login user
* @param workflowCreateRequest the new workflow object will be created
* @return New ProcessDefinition object created just now
*/
@@ -79,7 +79,7 @@ public interface ProcessDefinitionService {
/**
* query process definition list
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
* @return definition list
*/
@@ -89,7 +89,7 @@ public interface ProcessDefinitionService {
/**
* query process definition simple list
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
* @return definition simple list
*/
@@ -99,13 +99,13 @@ public interface ProcessDefinitionService {
/**
* query process definition list paging
*
- * @param loginUser login user
- * @param projectCode project code
- * @param searchVal search value
+ * @param loginUser login user
+ * @param projectCode project code
+ * @param searchVal search value
* @param otherParamsJson otherParamsJson handle other params
- * @param pageNo page number
- * @param pageSize page size
- * @param userId user id
+ * @param pageNo page number
+ * @param pageSize page size
+ * @param userId user id
* @return process definition page
*/
PageInfo<ProcessDefinition> queryProcessDefinitionListPaging(User loginUser,
@@ -119,7 +119,7 @@ public interface ProcessDefinitionService {
/**
* Filter resource process definitions
*
- * @param loginUser login user
+ * @param loginUser login user
* @param workflowFilterRequest workflow filter requests
* @return List process definition
*/
@@ -129,9 +129,9 @@ public interface ProcessDefinitionService {
/**
* query detail of process definition
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param code process definition code
+ * @param code process definition code
* @return process definition detail
*/
@@ -150,6 +150,7 @@ public interface ProcessDefinitionService {
long code);
Optional<ProcessDefinition> queryWorkflowDefinition(long workflowDefinitionCode, int workflowDefinitionVersion);
+
ProcessDefinition queryWorkflowDefinitionThrowExceptionIfNotFound(long workflowDefinitionCode,
int workflowDefinitionVersion);
@@ -169,9 +170,9 @@ public interface ProcessDefinitionService {
/**
* batch copy process definition
*
- * @param loginUser loginUser
- * @param projectCode projectCode
- * @param codes processDefinitionCodes
+ * @param loginUser loginUser
+ * @param projectCode projectCode
+ * @param codes processDefinitionCodes
* @param targetProjectCode targetProjectCode
*/
Map<String, Object> batchCopyProcessDefinition(User loginUser,
@@ -182,9 +183,9 @@ public interface ProcessDefinitionService {
/**
* batch move process definition
*
- * @param loginUser loginUser
- * @param projectCode projectCode
- * @param codes processDefinitionCodes
+ * @param loginUser loginUser
+ * @param projectCode projectCode
+ * @param codes processDefinitionCodes
* @param targetProjectCode targetProjectCode
*/
Map<String, Object> batchMoveProcessDefinition(User loginUser,
@@ -195,15 +196,15 @@ public interface ProcessDefinitionService {
/**
* update process definition, with whole process definition object including task definition, task relation and location.
*
- * @param loginUser login user
- * @param projectCode project code
- * @param name process definition name
- * @param code process definition code
- * @param description description
- * @param globalParams global params
- * @param locations locations for nodes
- * @param timeout timeout
- * @param taskRelationJson relation json for nodes
+ * @param loginUser login user
+ * @param projectCode project code
+ * @param name process definition name
+ * @param code process definition code
+ * @param description description
+ * @param globalParams global params
+ * @param locations locations for nodes
+ * @param timeout timeout
+ * @param taskRelationJson relation json for nodes
* @param taskDefinitionJson taskDefinitionJson
* @return update result code
*/
@@ -222,9 +223,9 @@ public interface ProcessDefinitionService {
/**
* verify process definition name unique
*
- * @param loginUser login user
- * @param projectCode project code
- * @param name name
+ * @param loginUser login user
+ * @param projectCode project code
+ * @param name name
* @param processDefinitionCode processDefinitionCode
* @return true if process definition name not exists, otherwise false
*/
@@ -236,9 +237,9 @@ public interface ProcessDefinitionService {
/**
* batch delete process definition by code
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param codes process definition codes
+ * @param codes process definition codes
* @return delete result code
*/
Map<String, Object> batchDeleteProcessDefinitionByCodes(User loginUser,
@@ -250,10 +251,10 @@ public interface ProcessDefinitionService {
/**
* batch export process definition by codes
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param codes process definition codes
- * @param response http servlet response
+ * @param codes process definition codes
+ * @param response http servlet response
*/
void batchExportProcessDefinitionByCodes(User loginUser,
long projectCode,
@@ -263,9 +264,9 @@ public interface ProcessDefinitionService {
/**
* import process definition
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param file process metadata json file
+ * @param file process metadata json file
* @return import process
*/
Map<String, Object> importProcessDefinition(User loginUser,
@@ -275,9 +276,9 @@ public interface ProcessDefinitionService {
/**
* import sql process definition
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param file sql file, zip
+ * @param file sql file, zip
* @return import process
*/
Map<String, Object> importSqlProcessDefinition(User loginUser,
@@ -296,9 +297,9 @@ public interface ProcessDefinitionService {
/**
* get task node details based on process definition
*
- * @param loginUser loginUser
+ * @param loginUser loginUser
* @param projectCode project code
- * @param code processDefinition code
+ * @param code processDefinition code
* @return task node list
*/
Map<String, Object> getTaskNodeListByDefinitionCode(User loginUser,
@@ -308,9 +309,9 @@ public interface ProcessDefinitionService {
/**
* get task node details map based on process definition
*
- * @param loginUser loginUser
+ * @param loginUser loginUser
* @param projectCode project code
- * @param codes define code list
+ * @param codes define code list
* @return task node list
*/
Map<String, Object> getNodeListMapByDefinitionCodes(User loginUser,
@@ -336,7 +337,7 @@ public interface ProcessDefinitionService {
/**
* query process definition list by project code
*
- * @param projectCode project code
+ * @param projectCode project code
* @param processDefinitionCode process definition code
* @return process definitions in the project
*/
@@ -346,8 +347,8 @@ public interface ProcessDefinitionService {
* Encapsulates the TreeView structure
*
* @param projectCode project code
- * @param code process definition code
- * @param limit limit
+ * @param code process definition code
+ * @param limit limit
* @return tree view json data
*/
Map<String, Object> viewTree(User loginUser, long projectCode, long code, Integer limit);
@@ -355,10 +356,10 @@ public interface ProcessDefinitionService {
/**
* switch the defined process definition version
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param code process definition code
- * @param version the version user want to switch
+ * @param code process definition code
+ * @param version the version user want to switch
* @return switch process definition version result code
*/
Map<String, Object> switchProcessDefinitionVersion(User loginUser,
@@ -369,11 +370,11 @@ public interface ProcessDefinitionService {
/**
* query the pagination versions info by one certain process definition code
*
- * @param loginUser login user info to check auth
+ * @param loginUser login user info to check auth
* @param projectCode project code
- * @param pageNo page number
- * @param pageSize page size
- * @param code process definition code
+ * @param pageNo page number
+ * @param pageSize page size
+ * @param code process definition code
* @return the pagination process definition versions info of the certain process definition
*/
Result queryProcessDefinitionVersions(User loginUser,
@@ -385,22 +386,21 @@ public interface ProcessDefinitionService {
/**
* delete one certain process definition by version number and process definition code
*
- * @param loginUser login user info to check auth
- * @param projectCode project code
- * @param code process definition code
- * @param version version number
- * @return delele result code
+ * @param loginUser login user info to check auth
+ * @param projectCode project code
+ * @param workflowDefinitionCode process definition code
+ * @param workflowDefinitionVersion version number
*/
- Map<String, Object> deleteProcessDefinitionVersion(User loginUser,
- long projectCode,
- long code,
- int version);
+ void deleteProcessDefinitionVersion(User loginUser,
+ long projectCode,
+ long workflowDefinitionCode,
+ int workflowDefinitionVersion);
/**
* update process definition basic info, not including task definition, task relation and location.
*
- * @param loginUser login user
- * @param workflowCode workflow resource code you want to update
+ * @param loginUser login user
+ * @param workflowCode workflow resource code you want to update
* @param workflowUpdateRequest workflow update requests
* @return ProcessDefinition instance
*/
@@ -420,9 +420,10 @@ public interface ProcessDefinitionService {
/**
* view process variables
- * @param loginUser login user
+ *
+ * @param loginUser login user
* @param projectCode project code
- * @param code process definition code
+ * @param code process definition code
* @return variables data
*/
Map<String, Object> viewVariables(User loginUser, long projectCode, long code);
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 ccb3f98b4d..7bf49e0016 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
@@ -49,9 +49,9 @@ public interface ProcessInstanceService {
/**
* query process instance by id
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param processId process instance id
+ * @param processId process instance id
* @return process instance detail
*/
Map<String, Object> queryProcessInstanceById(User loginUser,
@@ -73,17 +73,17 @@ public interface ProcessInstanceService {
/**
* paging query process instance list, filtering according to project, process definition, time range, keyword, process status
*
- * @param loginUser login user
- * @param projectCode project code
- * @param pageNo page number
- * @param pageSize page size
+ * @param loginUser login user
+ * @param projectCode project code
+ * @param pageNo page number
+ * @param pageSize page size
* @param processDefineCode process definition code
- * @param searchVal search value
- * @param stateType state type
- * @param host host
- * @param startDate start time
- * @param endDate end time
- * @param otherParamsJson otherParamsJson handle other params
+ * @param searchVal search value
+ * @param stateType state type
+ * @param host host
+ * @param startDate start time
+ * @param endDate end time
+ * @param otherParamsJson otherParamsJson handle other params
* @return process instance list
*/
Result<PageInfo<ProcessInstance>> queryProcessInstanceList(User loginUser,
@@ -102,7 +102,7 @@ public interface ProcessInstanceService {
/**
* paging query process instance list, filtering according to project, process definition, time range, keyword, process status
*
- * @param loginUser login user
+ * @param loginUser login user
* @param workflowInstanceQueryRequest workflowInstanceQueryRequest
* @return process instance list
*/
@@ -112,9 +112,9 @@ public interface ProcessInstanceService {
/**
* query task list by process instance id
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param processId process instance id
+ * @param processId process instance id
* @return task list for the process instance
* @throws IOException io exception
*/
@@ -127,9 +127,9 @@ public interface ProcessInstanceService {
/**
* query sub process instance detail info by task id
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param taskId task id
+ * @param taskId task id
* @return sub process instance detail
*/
Map<String, Object> querySubProcessInstanceByTaskId(User loginUser,
@@ -142,16 +142,16 @@ public interface ProcessInstanceService {
/**
* update process instance
*
- * @param loginUser login user
- * @param projectCode project code
- * @param taskRelationJson process task relation json
+ * @param loginUser login user
+ * @param projectCode project code
+ * @param taskRelationJson process task relation json
* @param taskDefinitionJson taskDefinitionJson
- * @param processInstanceId process instance id
- * @param scheduleTime schedule time
- * @param syncDefine sync define
- * @param globalParams global params
- * @param locations locations for nodes
- * @param timeout timeout
+ * @param processInstanceId process instance id
+ * @param scheduleTime schedule time
+ * @param syncDefine sync define
+ * @param globalParams global params
+ * @param locations locations for nodes
+ * @param timeout timeout
* @return update result code
*/
Map<String, Object> updateProcessInstance(User loginUser,
@@ -168,9 +168,9 @@ public interface ProcessInstanceService {
/**
* query parent process instance detail info by sub process instance id
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param subId sub process id
+ * @param subId sub process id
* @return parent instance detail
*/
Map<String, Object> queryParentInstanceBySubId(User loginUser,
@@ -190,7 +190,7 @@ public interface ProcessInstanceService {
/**
* view process instance variables
*
- * @param projectCode project code
+ * @param projectCode project code
* @param processInstanceId process instance id
* @return variables data
*/
@@ -199,7 +199,7 @@ public interface ProcessInstanceService {
/**
* encapsulation gantt structure
*
- * @param projectCode project code
+ * @param projectCode project code
* @param processInstanceId process instance id
* @return gantt tree data
* @throws Exception exception when json parse
@@ -210,17 +210,29 @@ public interface ProcessInstanceService {
* query process instance by processDefinitionCode and stateArray
*
* @param processDefinitionCode processDefinitionCode
- * @param states states array
+ * @param states states array
* @return process instance list
*/
List<ProcessInstance> queryByProcessDefineCodeAndStatus(Long processDefinitionCode,
int[] states);
+ /**
+ * query process instance by processDefinitionCode and stateArray
+ *
+ * @param workflowDefinitionCode workflowDefinitionCode
+ * @param workflowDefinitionVersion workflowDefinitionVersion
+ * @param states states array
+ * @return process instance list
+ */
+ List<ProcessInstance> queryByWorkflowCodeVersionStatus(Long workflowDefinitionCode,
+ int workflowDefinitionVersion,
+ int[] states);
+
/**
* query process instance by processDefinitionCode
*
* @param processDefinitionCode processDefinitionCode
- * @param size size
+ * @param size size
* @return process instance list
*/
List<ProcessInstance> queryByProcessDefineCode(Long processDefinitionCode,
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 173268c53a..09fe59e523 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
@@ -70,6 +70,7 @@ import org.apache.dolphinscheduler.common.enums.ProcessExecutionTypeEnum;
import org.apache.dolphinscheduler.common.enums.ReleaseState;
import org.apache.dolphinscheduler.common.enums.TimeoutFlag;
import org.apache.dolphinscheduler.common.enums.UserType;
+import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
import org.apache.dolphinscheduler.common.graph.DAG;
import org.apache.dolphinscheduler.common.lifecycle.ServerLifeCycleManager;
import org.apache.dolphinscheduler.common.model.TaskNodeRelation;
@@ -252,14 +253,14 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* create process definition
*
- * @param loginUser login user
- * @param projectCode project code
- * @param name process definition name
- * @param description description
- * @param globalParams global params
- * @param locations locations for nodes
- * @param timeout timeout
- * @param taskRelationJson relation json for nodes
+ * @param loginUser login user
+ * @param projectCode project code
+ * @param name process definition name
+ * @param description description
+ * @param globalParams global params
+ * @param locations locations for nodes
+ * @param timeout timeout
+ * @param taskRelationJson relation json for nodes
* @param taskDefinitionJson taskDefinitionJson
* @return create result code
*/
@@ -346,7 +347,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* create single process definition
*
- * @param loginUser login user
+ * @param loginUser login user
* @param workflowCreateRequest the new workflow object will be created
* @return New ProcessDefinition object created just now
*/
@@ -495,7 +496,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* query process definition list
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
* @return definition list
*/
@@ -518,7 +519,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* query process definition simple list
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
* @return definition simple list
*/
@@ -549,12 +550,12 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* query process definition list paging
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param searchVal search value
- * @param userId user id
- * @param pageNo page number
- * @param pageSize page size
+ * @param searchVal search value
+ * @param userId user id
+ * @param pageNo page number
+ * @param pageSize page size
* @return process definition page
*/
@Override
@@ -604,7 +605,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* Filter resource process definitions
*
- * @param loginUser login user
+ * @param loginUser login user
* @param workflowFilterRequest workflow filter requests
* @return List process definition
*/
@@ -644,9 +645,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* query detail of process definition
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param code process definition code
+ * @param code process definition code
* @return process definition detail
*/
@Override
@@ -675,7 +676,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
* query detail of process definition
*
* @param loginUser login user
- * @param code process definition code
+ * @param code process definition code
* @return process definition detail
*/
@Override
@@ -736,15 +737,15 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* update process definition
*
- * @param loginUser login user
- * @param projectCode project code
- * @param name process definition name
- * @param code process definition code
- * @param description description
- * @param globalParams global params
- * @param locations locations for nodes
- * @param timeout timeout
- * @param taskRelationJson relation json for nodes
+ * @param loginUser login user
+ * @param projectCode project code
+ * @param name process definition name
+ * @param code process definition code
+ * @param description description
+ * @param globalParams global params
+ * @param locations locations for nodes
+ * @param timeout timeout
+ * @param taskRelationJson relation json for nodes
* @param taskDefinitionJson taskDefinitionJson
* @return update result code
*/
@@ -817,11 +818,11 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* Task want to delete whether used in other task, should throw exception when have be used.
- *
+ * <p>
* This function avoid delete task already dependencies by other tasks by accident.
*
* @param processDefinition ProcessDefinition you change task definition and task relation
- * @param taskRelationList All the latest task relation list from process definition
+ * @param taskRelationList All the latest task relation list from process definition
*/
private void taskUsedInOtherTaskValid(ProcessDefinition processDefinition,
List<ProcessTaskRelationLog> taskRelationList) {
@@ -927,9 +928,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* verify process definition name unique
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param name name
+ * @param name name
* @return true if process definition name not exists, otherwise false
*/
@Override
@@ -999,7 +1000,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* Process definition want to delete whether used in other task, should throw exception when have be used.
- *
+ * <p>
* This function avoid delete process definition already dependencies by other tasks by accident.
*
* @param processDefinition ProcessDefinition you change task definition and task relation
@@ -1166,9 +1167,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* import process definition
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param file process metadata json file
+ * @param file process metadata json file
* @return import process
*/
@Override
@@ -1642,9 +1643,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* get task node details based on process definition
*
- * @param loginUser loginUser
+ * @param loginUser loginUser
* @param projectCode project code
- * @param code process definition code
+ * @param code process definition code
* @return task node list
*/
@Override
@@ -1671,9 +1672,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* get task node details map based on process definition
*
- * @param loginUser loginUser
+ * @param loginUser loginUser
* @param projectCode project code
- * @param codes define codes
+ * @param codes define codes
* @return task node list
*/
@Override
@@ -1724,7 +1725,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* query process definition all by project code
*
- * @param loginUser loginUser
+ * @param loginUser loginUser
* @param projectCode project code
* @return process definitions in the project
*/
@@ -1764,7 +1765,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* query process definition list by process definition code
*
- * @param projectCode project code
+ * @param projectCode project code
* @param processDefinitionCode process definition code
* @return task definition list in the process definition
*/
@@ -1800,8 +1801,8 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
* Encapsulates the TreeView structure
*
* @param projectCode project code
- * @param code process definition code
- * @param limit limit
+ * @param code process definition code
+ * @param limit limit
* @return tree view json data
*/
@Override
@@ -1963,9 +1964,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* batch copy process definition
*
- * @param loginUser loginUser
- * @param projectCode projectCode
- * @param codes processDefinitionCodes
+ * @param loginUser loginUser
+ * @param projectCode projectCode
+ * @param codes processDefinitionCodes
* @param targetProjectCode targetProjectCode
*/
@Override
@@ -1987,9 +1988,10 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* batch move process definition
* Will be deleted
- * @param loginUser loginUser
- * @param projectCode projectCode
- * @param codes processDefinitionCodes
+ *
+ * @param loginUser loginUser
+ * @param projectCode projectCode
+ * @param codes processDefinitionCodes
* @param targetProjectCode targetProjectCode
*/
@Override
@@ -2160,8 +2162,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* get new Task name or Process name when copy or import operate
+ *
* @param originalName Task or Process original name
- * @param suffix "_copy_" or "_import_"
+ * @param suffix "_copy_" or "_import_"
* @return new name
*/
public String getNewName(String originalName, String suffix) {
@@ -2183,10 +2186,10 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* switch the defined process definition version
*
- * @param loginUser login user
+ * @param loginUser login user
* @param projectCode project code
- * @param code process definition code
- * @param version the version user want to switch
+ * @param code process definition code
+ * @param version the version user want to switch
* @return switch process definition version result code
*/
@Override
@@ -2237,11 +2240,11 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* check batch operate result
*
- * @param srcProjectCode srcProjectCode
+ * @param srcProjectCode srcProjectCode
* @param targetProjectCode targetProjectCode
- * @param result result
+ * @param result result
* @param failedProcessList failedProcessList
- * @param isCopy isCopy
+ * @param isCopy isCopy
*/
private void checkBatchOperateResult(long srcProjectCode, long targetProjectCode,
Map<String, Object> result, List<String> failedProcessList, boolean isCopy) {
@@ -2268,11 +2271,11 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* query the pagination versions info by one certain process definition code
*
- * @param loginUser login user info to check auth
+ * @param loginUser login user info to check auth
* @param projectCode project code
- * @param pageNo page number
- * @param pageSize page size
- * @param code process definition code
+ * @param pageNo page number
+ * @param pageSize page size
+ * @param code process definition code
* @return the pagination process definition versions info of the certain process definition
*/
@Override
@@ -2304,52 +2307,42 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* delete one certain process definition by version number and process definition code
*
- * @param loginUser login user info to check auth
+ * @param loginUser login user info to check auth
* @param projectCode project code
- * @param code process definition code
- * @param version version number
- * @return delete result code
+ * @param code process definition code
+ * @param version version number
*/
@Override
@Transactional
- public Map<String, Object> deleteProcessDefinitionVersion(User loginUser, long projectCode, long code,
- int version) {
- Project project = projectMapper.queryByCode(projectCode);
- // check if user have write perm for project
- Map<String, Object> result = new HashMap<>();
- boolean hasProjectAndWritePerm = projectService.hasProjectAndWritePerm(loginUser, project, result);
- if (!hasProjectAndWritePerm) {
- return result;
- }
+ public void deleteProcessDefinitionVersion(User loginUser,
+ long projectCode,
+ long code,
+ int version) {
+ projectService.checkHasProjectWritePermissionThrowException(loginUser, projectCode);
ProcessDefinition processDefinition = processDefinitionMapper.queryByCode(code);
-
if (processDefinition == null || projectCode != processDefinition.getProjectCode()) {
- log.error("Process definition does not exist, code:{}.", code);
- putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, String.valueOf(code));
- } else {
- if (processDefinition.getVersion() == version) {
- log.warn(
- "Process definition can not be deleted due to version is being used, projectCode:{}, processDefinitionCode:{}, version:{}.",
- projectCode, code, version);
- putMsg(result, Status.MAIN_TABLE_USING_VERSION);
- return result;
- }
- int deleteLog = processDefinitionLogMapper.deleteByProcessDefinitionCodeAndVersion(code, version);
- int deleteRelationLog = processTaskRelationLogMapper.deleteByCode(code, version);
- if (deleteLog == 0 || deleteRelationLog == 0) {
- log.error(
- "Delete process definition version error, projectCode:{}, processDefinitionCode:{}, version:{}.",
- projectCode, code, version);
- putMsg(result, Status.DELETE_PROCESS_DEFINE_BY_CODE_ERROR);
- throw new ServiceException(Status.DELETE_PROCESS_DEFINE_BY_CODE_ERROR);
- }
- log.info(
- "Delete process definition version complete, projectCode:{}, processDefinitionCode:{}, version:{}.",
- projectCode, code, version);
- putMsg(result, Status.SUCCESS);
+ throw new ServiceException(Status.PROCESS_DEFINE_NOT_EXIST, code);
}
- return result;
+ if (processDefinition.getVersion() == version) {
+ log.warn("This version: {} of workflow: {} is the main version cannot delete by version", code, version);
+ throw new ServiceException(Status.MAIN_TABLE_USING_VERSION);
+ }
+ // check whether there exist running workflow instance under the process definition
+ List<ProcessInstance> workflowInstances = processInstanceService.queryByWorkflowCodeVersionStatus(
+ code,
+ version,
+ WorkflowExecutionStatus.getNotTerminalStatus());
+ if (CollectionUtils.isNotEmpty(workflowInstances)) {
+ throw new ServiceException(Status.DELETE_PROCESS_DEFINITION_EXECUTING_FAIL, workflowInstances.size());
+ }
+
+ int deleteLog = processDefinitionLogMapper.deleteByProcessDefinitionCodeAndVersion(code, version);
+ int deleteRelationLog = processTaskRelationLogMapper.deleteByCode(code, version);
+ if (deleteLog == 0 || deleteRelationLog == 0) {
+ throw new ServiceException(Status.DELETE_PROCESS_DEFINE_BY_CODE_ERROR);
+ }
+ log.info("Delete version: {} of workflow: {}, projectCode:{}.", version, code, projectCode);
}
private void updateWorkflowValid(User user, ProcessDefinition oldProcessDefinition,
@@ -2380,9 +2373,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* update single resource workflow
*
- * @param loginUser login user
- * @param workflowCode workflow resource code want to update
- * @param workflowUpdateRequest workflow update resource object
+ * @param loginUser login user
+ * @param workflowCode workflow resource code want to update
+ * @param workflowUpdateRequest workflow update resource object
* @return Process definition
*/
@Override
@@ -2546,7 +2539,8 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* view process variables
- * @param loginUser login user
+ *
+ * @param loginUser login user
* @param projectCode project code
* @param code process definition code
* @return variables data
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessInstanceServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessInstanceServiceImpl.java
index 899f578e57..36cc986607 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessInstanceServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessInstanceServiceImpl.java
@@ -1064,6 +1064,13 @@ public class ProcessInstanceServiceImpl extends BaseServiceImpl implements Proce
return processInstanceMapper.queryByProcessDefineCodeAndStatus(processDefinitionCode, states);
}
+ @Override
+ public List<ProcessInstance> queryByWorkflowCodeVersionStatus(Long workflowDefinitionCode,
+ int workflowDefinitionVersion, int[] states) {
+ return processInstanceDao.queryByWorkflowCodeVersionStatus(workflowDefinitionCode, workflowDefinitionVersion,
+ states);
+ }
+
/**
* query process instance by processDefinitionCode
*
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/ProcessDefinitionControllerTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/ProcessDefinitionControllerTest.java
index e9538667d0..91578c974b 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/ProcessDefinitionControllerTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/ProcessDefinitionControllerTest.java
@@ -17,6 +17,8 @@
package org.apache.dolphinscheduler.api.controller;
+import static org.mockito.Mockito.doNothing;
+
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.service.impl.ProcessDefinitionServiceImpl;
import org.apache.dolphinscheduler.api.utils.PageInfo;
@@ -183,7 +185,7 @@ public class ProcessDefinitionControllerTest {
Map<String, Object> result = new HashMap<>();
putMsg(result, Status.SUCCESS);
- Mockito.doNothing().when(processDefinitionService)
+ doNothing().when(processDefinitionService)
.offlineWorkflowDefinition(user, projectCode, id);
Result<Boolean> response =
processDefinitionController.releaseProcessDefinition(user, projectCode, id, ReleaseState.OFFLINE);
@@ -383,7 +385,7 @@ public class ProcessDefinitionControllerTest {
String processDefinitionIds = "1,2";
long projectCode = 1L;
HttpServletResponse response = new MockHttpServletResponse();
- Mockito.doNothing().when(this.processDefinitionService).batchExportProcessDefinitionByCodes(user, projectCode,
+ doNothing().when(this.processDefinitionService).batchExportProcessDefinitionByCodes(user, projectCode,
processDefinitionIds, response);
processDefinitionController.batchExportProcessDefinitionByCodes(user, projectCode, processDefinitionIds,
response);
@@ -420,13 +422,11 @@ public class ProcessDefinitionControllerTest {
@Test
public void testDeleteProcessDefinitionVersion() {
long projectCode = 1L;
- Map<String, Object> resultMap = new HashMap<>();
- putMsg(resultMap, Status.SUCCESS);
- Mockito.when(processDefinitionService.deleteProcessDefinitionVersion(
- user, projectCode, 1, 10)).thenReturn(resultMap);
- Result result = processDefinitionController.deleteProcessDefinitionVersion(
- user, projectCode, 1, 10);
- Assertions.assertEquals(Status.SUCCESS.getCode(), (int) result.getCode());
+ long workflowCode = 1L;
+ int workflowVersion = 10;
+ doNothing().when(processDefinitionService).deleteProcessDefinitionVersion(user, projectCode, workflowCode,
+ workflowVersion);
+ processDefinitionController.deleteProcessDefinitionVersion(user, projectCode, workflowCode, workflowVersion);
}
@Test
diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/WorkflowExecutionStatus.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/WorkflowExecutionStatus.java
index e8ec2dbc08..880db60257 100644
--- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/WorkflowExecutionStatus.java
+++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/WorkflowExecutionStatus.java
@@ -52,6 +52,17 @@ public enum WorkflowExecutionStatus {
READY_STOP.getCode()
};
+ private static final int[] NOT_TERMINAL_STATUS = new int[]{
+ SUBMITTED_SUCCESS.getCode(),
+ RUNNING_EXECUTION.getCode(),
+ DELAY_EXECUTION.getCode(),
+ READY_PAUSE.getCode(),
+ READY_STOP.getCode(),
+ SERIAL_WAIT.getCode(),
+ READY_BLOCK.getCode(),
+ WAIT_TO_RUN.getCode()
+ };
+
static {
for (WorkflowExecutionStatus executionStatus : WorkflowExecutionStatus.values()) {
CODE_MAP.put(executionStatus.getCode(), executionStatus);
@@ -116,6 +127,10 @@ public enum WorkflowExecutionStatus {
return NEED_FAILOVER_STATES;
}
+ public static int[] getNotTerminalStatus() {
+ return NOT_TERMINAL_STATUS;
+ }
+
@EnumValue
private final int code;
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.java
index a0a81374f5..20c259da9b 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.java
@@ -259,6 +259,10 @@ public interface ProcessInstanceMapper extends BaseMapper<ProcessInstance> {
List<ProcessInstance> queryByProcessDefineCodeAndStatus(@Param("processDefinitionCode") Long processDefinitionCode,
@Param("states") int[] states);
+ List<ProcessInstance> queryByWorkflowCodeVersionStatus(@Param("workflowDefinitionCode") long workflowDefinitionCode,
+ @Param("workflowDefinitionVersion") int workflowDefinitionVersion,
+ @Param("states") int[] states);
+
List<ProcessInstance> queryByProcessDefineCodeAndProcessDefinitionVersionAndStatusAndNextId(@Param("processDefinitionCode") Long processDefinitionCode,
@Param("processDefinitionVersion") int processDefinitionVersion,
@Param("states") int[] states,
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/ProcessInstanceDao.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/ProcessInstanceDao.java
index c1bceab120..02703fe0f3 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/ProcessInstanceDao.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/ProcessInstanceDao.java
@@ -20,7 +20,7 @@ package org.apache.dolphinscheduler.dao.repository;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.plugin.task.api.model.DateInterval;
-import org.apache.ibatis.annotations.Param;
+import java.util.List;
public interface ProcessInstanceDao extends IDao<ProcessInstance> {
@@ -64,7 +64,7 @@ public interface ProcessInstanceDao extends IDao<ProcessInstance> {
* @param definitionCode definitionCode
* @return process instance
*/
- ProcessInstance queryFirstScheduleProcessInstance(@Param("processDefinitionCode") Long definitionCode);
+ ProcessInstance queryFirstScheduleProcessInstance(Long definitionCode);
/**
* query first manual process instance
@@ -72,7 +72,11 @@ public interface ProcessInstanceDao extends IDao<ProcessInstance> {
* @param definitionCode definitionCode
* @return process instance
*/
- ProcessInstance queryFirstStartProcessInstance(@Param("processDefinitionCode") Long definitionCode);
+ ProcessInstance queryFirstStartProcessInstance(Long definitionCode);
ProcessInstance querySubProcessInstanceByParentId(Integer processInstanceId, Integer taskInstanceId);
+
+ List<ProcessInstance> queryByWorkflowCodeVersionStatus(Long workflowDefinitionCode,
+ int workflowDefinitionVersion,
+ int[] states);
}
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/ProcessInstanceDaoImpl.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/ProcessInstanceDaoImpl.java
index a5562f7a91..8ff93b3a9f 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/ProcessInstanceDaoImpl.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/ProcessInstanceDaoImpl.java
@@ -25,6 +25,8 @@ import org.apache.dolphinscheduler.dao.repository.BaseDao;
import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
import org.apache.dolphinscheduler.plugin.task.api.model.DateInterval;
+import java.util.List;
+
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
@@ -82,7 +84,7 @@ public class ProcessInstanceDaoImpl extends BaseDao<ProcessInstance, ProcessInst
* find last manual process instance interval
*
* @param definitionCode process definition code
- * @param taskCode taskCode
+ * @param taskCode taskCode
* @param dateInterval dateInterval
* @return process instance
*/
@@ -129,4 +131,12 @@ public class ProcessInstanceDaoImpl extends BaseDao<ProcessInstance, ProcessInst
processInstance = queryById(processInstanceMap.getProcessInstanceId());
return processInstance;
}
+
+ @Override
+ public List<ProcessInstance> queryByWorkflowCodeVersionStatus(Long workflowDefinitionCode,
+ int workflowDefinitionVersion,
+ int[] states) {
+ return mybatisMapper.queryByWorkflowCodeVersionStatus(workflowDefinitionCode, workflowDefinitionVersion,
+ states);
+ }
}
diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.xml
index baeb5f90a7..095e266ed5 100644
--- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.xml
+++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.xml
@@ -259,6 +259,21 @@
</if>
order by id asc
</select>
+
+ <select id="queryByWorkflowCodeVersionStatus" resultType="org.apache.dolphinscheduler.dao.entity.ProcessInstance">
+ select
+ <include refid="baseSql"/>
+ from t_ds_process_instance
+ where process_definition_code=#{workflowDefinitionCode}
+ and process_definition_version = #{workflowDefinitionVersion}
+ <if test="states != null and states.length != 0">
+ and state in
+ <foreach collection="states" item="i" open="(" close=")" separator=",">
+ #{i}
+ </foreach>
+ </if>
+ </select>
+
<select id="queryByProcessDefineCodeAndProcessDefinitionVersionAndStatusAndNextId"
resultType="org.apache.dolphinscheduler.dao.entity.ProcessInstance">
select
diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/repository/impl/ProcessInstanceDaoImplTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/repository/impl/ProcessInstanceDaoImplTest.java
new file mode 100644
index 0000000000..421309130f
--- /dev/null
+++ b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/repository/impl/ProcessInstanceDaoImplTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.repository.impl;
+
+import static org.apache.commons.collections4.CollectionUtils.isEmpty;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
+import org.apache.dolphinscheduler.dao.BaseDaoTest;
+import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
+import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+
+class ProcessInstanceDaoImplTest extends BaseDaoTest {
+
+ @Autowired
+ private ProcessInstanceDao processInstanceDao;
+
+ @Test
+ void queryByWorkflowCodeVersionStatus_EMPTY_INSTANCE() {
+ long workflowDefinitionCode = 1L;
+ int workflowDefinitionVersion = 1;
+ int[] status = WorkflowExecutionStatus.getNeedFailoverWorkflowInstanceState();
+
+ assertTrue(isEmpty(processInstanceDao.queryByWorkflowCodeVersionStatus(workflowDefinitionCode,
+ workflowDefinitionVersion, status)));
+ }
+
+ @Test
+ void queryByWorkflowCodeVersionStatus_EXIST_NOT_FINISH_INSTANCE() {
+ long workflowDefinitionCode = 1L;
+ int workflowDefinitionVersion = 1;
+ int[] status = WorkflowExecutionStatus.getNotTerminalStatus();
+
+ assertTrue(isEmpty(processInstanceDao.queryByWorkflowCodeVersionStatus(workflowDefinitionCode,
+ workflowDefinitionVersion, status)));
+
+ processInstanceDao.insert(createWorkflowInstance(workflowDefinitionCode, workflowDefinitionVersion,
+ WorkflowExecutionStatus.SUBMITTED_SUCCESS));
+ processInstanceDao.insert(createWorkflowInstance(workflowDefinitionCode, workflowDefinitionVersion,
+ WorkflowExecutionStatus.RUNNING_EXECUTION));
+ processInstanceDao.insert(createWorkflowInstance(workflowDefinitionCode, workflowDefinitionVersion,
+ WorkflowExecutionStatus.DELAY_EXECUTION));
+ processInstanceDao.insert(createWorkflowInstance(workflowDefinitionCode, workflowDefinitionVersion,
+ WorkflowExecutionStatus.READY_PAUSE));
+ processInstanceDao.insert(createWorkflowInstance(workflowDefinitionCode, workflowDefinitionVersion,
+ WorkflowExecutionStatus.READY_STOP));
+ processInstanceDao.insert(createWorkflowInstance(workflowDefinitionCode, workflowDefinitionVersion,
+ WorkflowExecutionStatus.SERIAL_WAIT));
+ processInstanceDao.insert(createWorkflowInstance(workflowDefinitionCode, workflowDefinitionVersion,
+ WorkflowExecutionStatus.READY_BLOCK));
+ processInstanceDao.insert(createWorkflowInstance(workflowDefinitionCode, workflowDefinitionVersion,
+ WorkflowExecutionStatus.WAIT_TO_RUN));
+ assertEquals(8, processInstanceDao
+ .queryByWorkflowCodeVersionStatus(workflowDefinitionCode, workflowDefinitionVersion, status).size());
+ }
+
+ @Test
+ void queryByWorkflowCodeVersionStatus_EXIST_FINISH_INSTANCE() {
+ long workflowDefinitionCode = 1L;
+ int workflowDefinitionVersion = 1;
+ int[] status = WorkflowExecutionStatus.getNotTerminalStatus();
+
+ processInstanceDao.insert(createWorkflowInstance(workflowDefinitionCode, workflowDefinitionVersion,
+ WorkflowExecutionStatus.PAUSE));
+ processInstanceDao.insert(createWorkflowInstance(workflowDefinitionCode, workflowDefinitionVersion,
+ WorkflowExecutionStatus.STOP));
+ processInstanceDao.insert(createWorkflowInstance(workflowDefinitionCode, workflowDefinitionVersion,
+ WorkflowExecutionStatus.FAILURE));
+ processInstanceDao.insert(createWorkflowInstance(workflowDefinitionCode, workflowDefinitionVersion,
+ WorkflowExecutionStatus.SUCCESS));
+ assertTrue(isEmpty(processInstanceDao.queryByWorkflowCodeVersionStatus(workflowDefinitionCode,
+ workflowDefinitionVersion, status)));
+ }
+
+ private ProcessInstance createWorkflowInstance(Long workflowDefinitionCode, int workflowDefinitionVersion,
+ WorkflowExecutionStatus status) {
+ ProcessInstance processInstance = new ProcessInstance();
+ processInstance.setName("WorkflowInstance" + System.currentTimeMillis());
+ processInstance.setProcessDefinitionCode(workflowDefinitionCode);
+ processInstance.setProcessDefinitionVersion(workflowDefinitionVersion);
+ processInstance.setState(status);
+ return processInstance;
+ }
+
+}