[FIX] Completed tasks cannot be re-executed in a workflow instance (#15884)

* fix bug: Failed to resume stopped workflow instance

* Revert "fix bug: Failed to resume stopped workflow instance"

This reverts commit 1546e9d5a51178a94bedd18a718b15431355428b.

* fix bug : Completed tasks cannot be re-executed in a workflow instance

---------

Co-authored-by: Rick Cheng <rickchengx@gmail.com>

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java

@@ -2120,13 +2120,9 @@ public class WorkflowExecuteRunnable implements IWorkflowExecuteRunnable {
         workflowInstance.setVarPool(JSONUtils.toJsonString(processProperties));
         processInstanceDao.updateById(workflowInstance);
 
-        // remove task instance from taskInstanceMap, completeTaskSet, validTaskMap, errorTaskMap
+        // remove task instance from taskInstanceMap,taskCodeInstanceMap , completeTaskSet, validTaskMap, errorTaskMap
-        // completeTaskSet remove dependency taskInstanceMap, so the sort can't change
+        completeTaskSet.removeIf(dag::containsNode);
-        completeTaskSet.removeIf(taskCode -> {
+        taskCodeInstanceMap.entrySet().removeIf(entity -> dag.containsNode(entity.getValue().getTaskCode()));
-            Optional<TaskInstance> existTaskInstanceOptional = getTaskInstance(taskCode);
-            return existTaskInstanceOptional
-                    .filter(taskInstance -> dag.containsNode(taskInstance.getTaskCode())).isPresent();
-        });
         taskInstanceMap.entrySet().removeIf(entry -> dag.containsNode(entry.getValue().getTaskCode()));
         validTaskMap.entrySet().removeIf(entry -> dag.containsNode(entry.getKey()));
         errorTaskMap.entrySet().removeIf(entry -> dag.containsNode(entry.getKey()));