private Optional<TaskModel> findChangedSubWorkflowTask(WorkflowModel workflow) {
        WorkflowDef workflowDef =
                Optional.ofNullable(workflow.getWorkflowDefinition())
                        .orElseGet(
                                () ->
                                        metadataDAO
                                                .getWorkflowDef(
                                                        workflow.getWorkflowName(),
                                                        workflow.getWorkflowVersion())
                                                .orElseThrow(
                                                        () ->
                                                                new TransientException(
                                                                        "Workflow Definition is not found")));
        if (workflowDef.containsType(TaskType.TASK_TYPE_SUB_WORKFLOW)
                || workflow.getWorkflowDefinition()
                        .containsType(TaskType.TASK_TYPE_FORK_JOIN_DYNAMIC)) {
            return workflow.getTasks().stream()
                    .filter(
                            t ->
                                    t.getTaskType().equals(TaskType.TASK_TYPE_SUB_WORKFLOW)
                                            && t.isSubworkflowChanged()
                                            && !t.isRetried())
                    .findFirst();
        }
        return Optional.empty();
    }
@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