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 testRestartDirectlyWhileForeachStepRunning() {
    stepInstance.getRuntimeState().setStatus(StepInstance.Status.RUNNING);
    ((TypedStep) stepInstance.getDefinition()).setType(StepType.FOREACH);
    ForeachArtifact artifact = new ForeachArtifact();
    artifact.setForeachWorkflowId("maestro-foreach-wf");
    artifact.setNextLoopIndex(12);
    artifact.setForeachOverview(new ForeachStepOverview());
    artifact.getForeachOverview().addOne(10, WorkflowInstance.Status.FAILED, null);
    artifact.getForeachOverview().refreshDetail();
    stepInstance.setArtifacts(Collections.singletonMap(Artifact.Type.FOREACH.key(), artifact));
    stepInstanceDao.insertOrUpsertStepInstance(stepInstance, true);
    RunResponse restartStepInfo = RunResponse.builder().instance(instance).stepId("job1").build();
    RunRequest runRequest =
        RunRequest.builder()
            .requester(user)
            .currentPolicy(RunPolicy.RESTART_FROM_SPECIFIC)
            .restartConfig(
                RestartConfig.builder()
                    .addRestartNode("maestro-foreach-wf", 10, "job2")
                    .addRestartNode("sample-dag-test-3", 1, "job1")
                    .build())
            .build();

    RunResponse response = actionDao.restartDirectly(restartStepInfo, runRequest, false);

    Assert.assertEquals("sample-dag-test-3", response.getWorkflowId());
    Assert.assertEquals(1, response.getWorkflowInstanceId());
    Assert.assertEquals(1, response.getWorkflowRunId());
    Assert.assertEquals("job1", response.getStepId());
    Assert.assertEquals(
        "User [tester] take action [RESTART] on the step",
        response.getTimelineEvent().getMessage());

    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(restartStepInfo, runRequest, false));

    artifact.setNextLoopIndex(9);
    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 its status [RUNNING] ",
        () -> actionDao.restartDirectly(restartStepInfo, runRequest, false));

    artifact.setNextLoopIndex(10);
    artifact.setPendingAction(ForeachAction.builder().build());
    stepInstanceDao.insertOrUpsertStepInstance(stepInstance, true);
    AssertHelper.assertThrows(
        "Cannot manually RESTART the step",
        MaestroResourceConflictException.class,
        "The foreach iteration [10] is not ready to be restarted for foreach step ",
        () -> actionDao.restartDirectly(restartStepInfo, runRequest, false));
  }