public void restart(String workflowId, boolean useLatestDefinitions) {
        final WorkflowModel workflow = executionDAOFacade.getWorkflowModel(workflowId, true);

        if (!workflow.getStatus().isTerminal()) {
            String errorMsg =
                    String.format(
                            "Workflow: %s is not in terminal state, unable to restart.", workflow);
            LOGGER.error(errorMsg);
            throw new ConflictException(errorMsg);
        }

        WorkflowDef workflowDef;
        if (useLatestDefinitions) {
            workflowDef =
                    metadataDAO
                            .getLatestWorkflowDef(workflow.getWorkflowName())
                            .orElseThrow(
                                    () ->
                                            new NotFoundException(
                                                    "Unable to find latest definition for %s",
                                                    workflowId));
            workflow.setWorkflowDefinition(workflowDef);
            workflowDef = metadataMapperService.populateTaskDefinitions(workflowDef);
        } else {
            workflowDef =
                    Optional.ofNullable(workflow.getWorkflowDefinition())
                            .orElseGet(
                                    () ->
                                            metadataDAO
                                                    .getWorkflowDef(
                                                            workflow.getWorkflowName(),
                                                            workflow.getWorkflowVersion())
                                                    .orElseThrow(
                                                            () ->
                                                                    new NotFoundException(
                                                                            "Unable to find definition for %s",
                                                                            workflowId)));
        }

        if (!workflowDef.isRestartable()
                && workflow.getStatus()
                        .equals(
                                WorkflowModel.Status
                                        .COMPLETED)) { // Can only restart non-completed workflows
            // when the configuration is set to false
            throw new NotFoundException("Workflow: %s is non-restartable", workflow);
        }

        // Reset the workflow in the primary datastore and remove from indexer; then re-create it
        executionDAOFacade.resetWorkflow(workflowId);

        workflow.getTasks().clear();
        workflow.setReasonForIncompletion(null);
        workflow.setFailedTaskId(null);
        workflow.setCreateTime(System.currentTimeMillis());
        workflow.setEndTime(0);
        workflow.setLastRetriedTime(0);
        // Change the status to running
        workflow.setStatus(WorkflowModel.Status.RUNNING);
        workflow.setOutput(null);
        workflow.setExternalOutputPayloadStoragePath(null);

        try {
            executionDAOFacade.createWorkflow(workflow);
        } catch (Exception e) {
            Monitors.recordWorkflowStartError(
                    workflowDef.getName(), WorkflowContext.get().getClientApp());
            LOGGER.error("Unable to restart workflow: {}", workflowDef.getName(), e);
            terminateWorkflow(workflowId, "Error when restarting the workflow");
            throw e;
        }

        metadataMapperService.populateWorkflowWithDefinitions(workflow);
        decide(workflowId);

        updateAndPushParents(workflow, "restarted");
    }
@param workflowId the id of the workflow to be restarted
@param useLatestDefinitions if true, use the latest workflow and task definitions upon
    restart
@throws ConflictException Workflow is not in a terminal state.
@throws NotFoundException Workflow definition is not found or Workflow is deemed
    non-restartable as per workflow definition.
core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutor.java