public RunResponse restartDirectly(
      RunResponse restartStepInfo, RunRequest runRequest, boolean blocking) {
    WorkflowInstance instance = restartStepInfo.getInstance();
    String stepId = restartStepInfo.getStepId();
    validateStepId(instance, stepId, Actions.StepInstanceAction.RESTART);
    StepInstance stepInstance =
        getStepInstanceAndValidate(instance, stepId, runRequest.getRestartConfig());
    // prepare payload and then add to db
    StepAction stepAction = StepAction.createRestart(stepInstance, runRequest);
    saveAction(stepInstance, stepAction);
    if (blocking) {
      return waitResponseWithTimeout(stepInstance, stepAction);
    } else {
      return RunResponse.from(stepInstance, stepAction.toTimelineEvent());
    }
  }
@Test
  public void testInvalidRestart() {
    RunResponse restartStepInfo1 =
        RunResponse.builder().instance(instance).stepId("not-existing").build();
    RunRequest runRequest =
        RunRequest.builder()
            .requester(user)
            .currentPolicy(RunPolicy.RESTART_FROM_SPECIFIC)
            .runParams(Collections.emptyMap())
            .build();

    AssertHelper.assertThrows(
        "Cannot manually RESTART the step",
        MaestroBadRequestException.class,
        "Cannot manually RESTART the step [not-existing] because the latest workflow run",
        () -> actionDao.restartDirectly(restartStepInfo1, runRequest, true));

    RunResponse restartStepInfo2 = RunResponse.builder().instance(instance).stepId("job.2").build();
    AssertHelper.assertThrows(
        "Cannot manually RESTART the step",
        MaestroInvalidStatusException.class,
        "Cannot manually RESTART the step [[sample-dag-test-3][1][1]][job.2] as it does not exist",
        () -> actionDao.restartDirectly(restartStepInfo2, runRequest, true));

    RunResponse restartStepInfo3 = RunResponse.builder().instance(instance).stepId("job1").build();
    AssertHelper.assertThrows(
        "Cannot manually RESTART the step",
        MaestroInvalidStatusException.class,
        "Cannot manually RESTART the step [sample-dag-test-3][1][1][job1] as its status [RUNNING] is either non-terminal or complete "
            + "while the workflow instance or its ancestor workflow instances is still non-terminal.",
        () -> actionDao.restartDirectly(restartStepInfo3, runRequest, true));

    RunResponse restartStepInfo4 = RunResponse.builder().instance(instance).stepId("job3").build();
    AssertHelper.assertThrows(
        "Cannot manually RESTART the step",
        MaestroNotFoundException.class,
        "step instance [job3][LATEST] not found (either not created or deleted)",
        () -> actionDao.restartDirectly(restartStepInfo4, runRequest, true));

    stepInstance.getRuntimeState().setStatus(StepInstance.Status.USER_FAILED);
    stepInstanceDao.insertOrUpsertStepInstance(stepInstance, true);

    AssertHelper.assertThrows(
        "Cannot manually RESTART the step",
        MaestroInvalidStatusException.class,
        "Cannot manually RESTART the step [sample-dag-test-3][1][1][job1] as the system is still retrying",
        () -> actionDao.restartDirectly(restartStepInfo3, runRequest, true));

    stepInstance.getRuntimeState().setStatus(StepInstance.Status.FATALLY_FAILED);
    stepInstance.getStepRetry().setRetryable(false);
    stepInstanceDao.insertOrUpsertStepInstance(stepInstance, true);

    AssertHelper.assertThrows(
        "Cannot manually RESTART the step",
        MaestroInvalidStatusException.class,
        "Cannot manually RESTART the step [sample-dag-test-3][1][1][job1] as step failure mode [IGNORE_FAILURE]",
        () -> actionDao.restartDirectly(restartStepInfo3, runRequest, true));

    ((TypedStep) stepInstance.getDefinition()).setFailureMode(FailureMode.FAIL_AFTER_RUNNING);
    stepInstanceDao.insertOrUpsertStepInstance(stepInstance, true);
    actionDao.restartDirectly(restartStepInfo3, runRequest, false);

    AssertHelper.assertThrows(
        "Cannot manually RESTART the step",
        MaestroResourceConflictException.class,
        "There is an ongoing action for this step [sample-dag-test-3][1][1][job1]",
        () -> actionDao.restartDirectly(restartStepInfo3, runRequest, true));
  }