Lazy evaluate workflow only in case of task is successful.

core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutor.java

@@ -925,10 +925,12 @@ boolean isLazyEvaluateWorkflow(WorkflowDef workflowDef, TaskModel task) {
                         .collect(Collectors.toList());
 
         if (forkTasks.stream().anyMatch(fork -> fork.has(taskRefName))) {
-            return joinTasks.stream().anyMatch(join -> join.getJoinOn().contains(taskRefName));
+            return joinTasks.stream().anyMatch(join -> join.getJoinOn().contains(taskRefName))
+                    && task.getStatus().isSuccessful();
         }
 
-        return workflowTasks.stream().noneMatch(t -> t.getTaskReferenceName().equals(taskRefName));
+        return workflowTasks.stream().noneMatch(t -> t.getTaskReferenceName().equals(taskRefName))
+                && task.getStatus().isSuccessful();
     }
 
     public TaskModel getTask(String taskId) {

core/src/test/java/com/netflix/conductor/core/execution/TestWorkflowExecutor.java

@@ -2487,6 +2487,7 @@ public void testIsLazyEvaluateWorkflow() {
         workflowDef.getTasks().addAll(List.of(simpleTask, forkTask, joinTask, doWhile));
 
         TaskModel task = new TaskModel();
+        task.setStatus(TaskModel.Status.COMPLETED);
 
         // when:
         task.setReferenceTaskName("dynamic");
@@ -2504,6 +2505,10 @@ public void testIsLazyEvaluateWorkflow() {
         task.setReferenceTaskName("loopTask__1");
         task.setIteration(1);
         assertFalse(workflowExecutor.isLazyEvaluateWorkflow(workflowDef, task));
+
+        task.setReferenceTaskName("branchTask1");
+        task.setStatus(TaskModel.Status.FAILED);
+        assertFalse(workflowExecutor.isLazyEvaluateWorkflow(workflowDef, task));
     }
 
     @Test