@VisibleForTesting
    List<TaskModel> dedupAndAddTasks(WorkflowModel workflow, List<TaskModel> tasks) {
        Set<String> tasksInWorkflow =
                workflow.getTasks().stream()
                        .map(task -> task.getReferenceTaskName() + "_" + task.getRetryCount())
                        .collect(Collectors.toSet());

        List<TaskModel> dedupedTasks =
                tasks.stream()
                        .filter(
                                task ->
                                        !tasksInWorkflow.contains(
                                                task.getReferenceTaskName()
                                                        + "_"
                                                        + task.getRetryCount()))
                        .collect(Collectors.toList());

        workflow.getTasks().addAll(dedupedTasks);
        return dedupedTasks;
    }
@param workflow the workflow to evaluate the state for
@return true if the workflow has completed (success or failed), false otherwise. Note: This
    method does not acquire the lock on the workflow and should ony be called / overridden if
    No locking is required or lock is acquired externally
core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutor.java