private void adjustStateIfSubWorkflowChanged(WorkflowModel workflow) {
        Optional<TaskModel> changedSubWorkflowTask = findChangedSubWorkflowTask(workflow);
        if (changedSubWorkflowTask.isPresent()) {
            // reset the flag
            TaskModel subWorkflowTask = changedSubWorkflowTask.get();
            subWorkflowTask.setSubworkflowChanged(false);
            executionDAOFacade.updateTask(subWorkflowTask);

            LOGGER.info(
                    "{} reset subworkflowChanged flag for {}",
                    workflow.toShortString(),
                    subWorkflowTask.getTaskId());

            // find all terminal and unsuccessful JOIN tasks and set them to IN_PROGRESS
            if (workflow.getWorkflowDefinition().containsType(TaskType.TASK_TYPE_JOIN)
                    || workflow.getWorkflowDefinition()
                            .containsType(TaskType.TASK_TYPE_FORK_JOIN_DYNAMIC)) {
                // if we are here, then the SUB_WORKFLOW task could be part of a FORK_JOIN or
                // FORK_JOIN_DYNAMIC
                // and the JOIN task(s) needs to be evaluated again, set them to IN_PROGRESS
                workflow.getTasks().stream()
                        .filter(UNSUCCESSFUL_JOIN_TASK)
                        .peek(
                                task -> {
                                    task.setStatus(TaskModel.Status.IN_PROGRESS);
                                    addTaskToQueue(task);
                                })
                        .forEach(executionDAOFacade::updateTask);
            }
        }
    }
@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