fix bug about CONDITION-task

1. condition shoudn't be in error list
2. condition need more checks when rusume_from_forced_success

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/DagHelper.java

@@ -22,6 +22,7 @@ import org.apache.dolphinscheduler.common.graph.DAG;
 import org.apache.dolphinscheduler.common.model.TaskNode;
 import org.apache.dolphinscheduler.common.model.TaskNodeRelation;
 import org.apache.dolphinscheduler.common.process.ProcessDag;
+import org.apache.dolphinscheduler.common.task.conditions.ConditionsParameters;
 import org.apache.dolphinscheduler.common.utils.CollectionUtils;
 import org.apache.dolphinscheduler.common.utils.JSONUtils;
 import org.apache.dolphinscheduler.common.utils.StringUtils;
@@ -275,7 +276,19 @@ public class DagHelper {
         }
         Collection<String> startVertexs = null;
         if (StringUtils.isNotEmpty(parentNodeName)) {
+            TaskNode task = dag.getNode(parentNodeName);
+            if (task.isConditionsTask() && completeTaskList.containsKey(parentNodeName)) {
+                ConditionsParameters conditionsParameters = JSONUtils.parseObject(task.getConditionResult(), ConditionsParameters.class);
+                TaskInstance taskInstance = completeTaskList.get(parentNodeName);
+                if (taskInstance.getState().typeIsSuccess()) {
+                    startVertexs = conditionsParameters.getSuccessNode();
+                } else if (taskInstance.getState().typeIsFailure()) {
+                    startVertexs = conditionsParameters.getFailedNode();
+                }
+            }
+            else {
                 startVertexs = dag.getSubsequentNodes(parentNodeName);
+            }
         } else {
             startVertexs = dag.getBeginNode();
         }

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

@@ -405,7 +405,7 @@ public class MasterExecThread implements Runnable {
             if(task.isTaskComplete()){
                 completeTaskList.put(task.getName(), task);
             }
-            if(task.getState().typeIsFailure() && !task.taskCanRetry()){
+            if (task.getState().typeIsFailure() && !task.taskCanRetry() && !task.isConditionsTask()) {
                 errorTaskList.put(task.getName(), task);
             }
         }