[Feature][API] process definition delete api add check process instance is already running. #3581 (#3582)

* process definition delete api add check process instance is already running.

* add checkstyle.

* processinstancemapper replaces processinstanceservice.

* processinstancemapper replaces processinstanceservice.

Co-authored-by: unknown <zhuang.chong@zhaopin.com.cn>

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java

@@ -193,7 +193,7 @@ public enum Status {
     BATCH_COPY_PROCESS_DEFINITION_ERROR(10159, "batch copy process definition error", "复制工作流错误"),
     BATCH_MOVE_PROCESS_DEFINITION_ERROR(10160, "batch move process definition error", "移动工作流错误"),
     QUERY_WORKFLOW_LINEAGE_ERROR(10161, "query workflow lineage error", "查询血缘失败"),
-
+    DELETE_PROCESS_DEFINITION_BY_ID_FAIL(10162,"delete process definition by id fail, for there are {0} process instances in executing using it", "删除工作流定义失败，有[{0}]个运行中的工作流实例正在使用"),
 
     UDF_FUNCTION_NOT_EXIST(20001, "UDF function not found", "UDF函数不存在"),
     UDF_FUNCTION_EXISTS(20002, "UDF function already exists", "UDF函数已存在"),
@@ -294,4 +294,4 @@ public enum Status {
             return this.enMsg;
         }
     }
-}
+}

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessInstanceService.java

@@ -24,7 +24,6 @@ import static org.apache.dolphinscheduler.common.Constants.LOCAL_PARAMS;
 import static org.apache.dolphinscheduler.common.Constants.PROCESS_INSTANCE_STATE;
 import static org.apache.dolphinscheduler.common.Constants.TASK_LIST;
 
-
 import org.apache.dolphinscheduler.api.dto.gantt.GanttDto;
 import org.apache.dolphinscheduler.api.dto.gantt.Task;
 import org.apache.dolphinscheduler.api.enums.Status;
@@ -711,4 +710,24 @@ public class ProcessInstanceService extends BaseService {
         return DagHelper.buildDagGraph(processDag);
     }
 
+    /**
+     * query process instance by processDefinitionId and stateArray
+     * @param processDefinitionId processDefinitionId
+     * @param states states array
+     * @return process instance list
+     */
+    public List<ProcessInstance> queryByProcessDefineIdAndStatus(int processDefinitionId, int[] states) {
+        return processInstanceMapper.queryByProcessDefineIdAndStatus(processDefinitionId, states);
+    }
+
+    /**
+     * query process instance by processDefinitionId
+     * @param processDefinitionId processDefinitionId
+     * @param size size
+     * @return process instance list
+     */
+    public List<ProcessInstance> queryByProcessDefineId(int processDefinitionId,int size) {
+        return processInstanceMapper.queryByProcessDefineId(processDefinitionId, size);
+    }
+
 }

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java

@@ -26,6 +26,7 @@ import org.apache.dolphinscheduler.api.enums.Status;
 import org.apache.dolphinscheduler.api.service.BaseService;
 import org.apache.dolphinscheduler.api.service.ProcessDefinitionService;
 import org.apache.dolphinscheduler.api.service.ProcessDefinitionVersionService;
+import org.apache.dolphinscheduler.api.service.ProcessInstanceService;
 import org.apache.dolphinscheduler.api.service.ProjectService;
 import org.apache.dolphinscheduler.api.service.SchedulerService;
 import org.apache.dolphinscheduler.api.utils.CheckUtils;
@@ -64,7 +65,6 @@ import org.apache.dolphinscheduler.dao.entity.Schedule;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.dao.entity.User;
 import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
-import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
 import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
 import org.apache.dolphinscheduler.dao.mapper.ScheduleMapper;
 import org.apache.dolphinscheduler.dao.mapper.TaskInstanceMapper;
@@ -134,7 +134,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
     private ProcessDefinitionMapper processDefineMapper;
 
     @Autowired
-    private ProcessInstanceMapper processInstanceMapper;
+    private ProcessInstanceService processInstanceService;
 
     @Autowired
     private TaskInstanceMapper taskInstanceMapper;
@@ -488,6 +488,12 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
             putMsg(result, Status.PROCESS_DEFINE_STATE_ONLINE, processDefinitionId);
             return result;
         }
+        // check process instances is already running
+        List<ProcessInstance> processInstances =  processInstanceService.queryByProcessDefineIdAndStatus(processDefinitionId, Constants.NOT_TERMINATED_STATES);
+        if (CollectionUtils.isNotEmpty(processInstances)) {
+            putMsg(result, Status.DELETE_PROCESS_DEFINITION_BY_ID_FAIL,processInstances.size());
+            return result;
+        }
 
         // get the timing according to the process definition
         List<Schedule> schedules = scheduleMapper.queryByProcessDefinitionId(processDefinitionId);
@@ -1247,7 +1253,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
         /**
          * List of process instances
          */
-        List<ProcessInstance> processInstanceList = processInstanceMapper.queryByProcessDefineId(processId, limit);
+        List<ProcessInstance> processInstanceList = processInstanceService.queryByProcessDefineId(processId, limit);
 
         for (ProcessInstance processInstance : processInstanceList) {
             processInstance.setDuration(DateUtils.differSec(processInstance.getStartTime(), processInstance.getEndTime()));

dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionServiceTest.java

@@ -41,7 +41,6 @@ import org.apache.dolphinscheduler.dao.entity.Schedule;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.dao.entity.User;
 import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
-import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
 import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
 import org.apache.dolphinscheduler.dao.mapper.ScheduleMapper;
 import org.apache.dolphinscheduler.dao.mapper.TaskInstanceMapper;
@@ -94,7 +93,7 @@ public class ProcessDefinitionServiceTest {
     private ProcessService processService;
 
     @Mock
-    private ProcessInstanceMapper processInstanceMapper;
+    private ProcessInstanceService processInstanceService;
 
     @Mock
     private TaskInstanceMapper taskInstanceMapper;
@@ -734,7 +733,7 @@ public class ProcessDefinitionServiceTest {
 
         //task instance not exist
         Mockito.when(processDefineMapper.selectById(46)).thenReturn(processDefinition);
-        Mockito.when(processInstanceMapper.queryByProcessDefineId(46, 10)).thenReturn(processInstanceList);
+        Mockito.when(processInstanceService.queryByProcessDefineId(46, 10)).thenReturn(processInstanceList);
         Mockito.when(taskInstanceMapper.queryByInstanceIdAndName(processInstance.getId(), "shell-1")).thenReturn(null);
         Map<String, Object> taskNullRes = processDefinitionService.viewTree(46, 10);
         Assert.assertEquals(Status.SUCCESS, taskNullRes.get(Constants.STATUS));

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.java

@@ -14,19 +14,22 @@
  * 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.ExecutionStatus;
 import org.apache.dolphinscheduler.dao.entity.ExecuteStatusCount;
 import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
-import com.baomidou.mybatisplus.core.mapper.BaseMapper;
-import com.baomidou.mybatisplus.core.metadata.IPage;
-import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
-import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
+
 import org.apache.ibatis.annotations.Param;
 
 import java.util.Date;
 import java.util.List;
 
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+
 /**
  * process instance mapper interface
  */
@@ -201,9 +204,20 @@ public interface ProcessInstanceMapper extends BaseMapper<ProcessInstance> {
      * @param endTime
      * @return ProcessInstance list
      */
+
     List<ProcessInstance> queryTopNProcessInstance(@Param("size") int size,
                                                    @Param("startTime") Date startTime,
                                                    @Param("endTime") Date endTime,
                                                    @Param("status")ExecutionStatus status);
+    /**
+     * query process instance by processDefinitionId and stateArray
+     * @param processDefinitionId processDefinitionId
+     * @param states states array
+     * @return process instance list
+     */
+
+    List<ProcessInstance> queryByProcessDefineIdAndStatus(
+            @Param("processDefinitionId") int processDefinitionId,
+            @Param("states") int[] states);
 
 }

dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.xml

@@ -191,6 +191,16 @@
         </if>
         order by end_time desc limit 1
     </select>
-
+    <select id="queryByProcessDefineIdAndStatus"
+            resultType="org.apache.dolphinscheduler.dao.entity.ProcessInstance">
+        select *
+        from t_ds_process_instance
+        where process_definition_id=#{processDefinitionId}
+        and state in
+        <foreach collection="states" item="i" open="(" close=")" separator=",">
+            #{i}
+        </foreach>
+        order by id asc
+    </select>
 
 </mapper>