[Fix-17030] Fix one task might be duplicate trigger if exist multiple predecessors (#17038)

Author: ruanwenjun

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/IWorkflowExecutionGraph.java

@@ -115,7 +115,8 @@ public interface IWorkflowExecutionGraph {
 
     /**
      * Check whether the given task can be trigger now.
-     * <p> The task can be trigger if all the predecessors are finished and all predecessors are not failure/pause/kill.
+     * <p> The task can be trigger only all the predecessors are finished and all predecessors are not failure/pause/kill.
+     * <p> Once the task has been triggered, then will also return false.
      */
     boolean isTriggerConditionMet(final ITaskExecutionRunnable taskExecutionRunnable);
 

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowExecutionGraph.java

@@ -181,6 +181,10 @@ public List<ITaskExecutionRunnable> getAllTaskExecutionRunnable() {
 
     @Override
     public boolean isTriggerConditionMet(final ITaskExecutionRunnable taskExecutionRunnable) {
+        if (isTaskExecutionRunnableActive(taskExecutionRunnable)
+                || isTaskExecutionRunnableInActive(taskExecutionRunnable)) {
+            return false;
+        }
         return getPredecessors(taskExecutionRunnable.getName())
                 .stream()
                 .allMatch(predecessor -> isTaskExecutionRunnableInActive(predecessor)