public RunResponse restart(RunRequest runRequest) {
    RunResponse runResponse = restartRecursively(runRequest);
    if (runResponse.getStatus() == RunResponse.Status.NON_TERMINAL_ERROR) {
      LOG.error(
          "workflow instance {} does not support restart action as it is in a non-terminal status [{}]",
          runRequest.getWorkflowIdentity(),
          runResponse.getTimelineEvent().getMessage());
      throw new MaestroBadRequestException(
          Collections.emptyList(),
          "workflow instance %s does not support restart action as it is in a non-terminal status [%s]",
          runRequest.getWorkflowIdentity(),
          runResponse.getTimelineEvent().getMessage());
    }
    return runResponse;
  }
@Test
  public void testRestartDelegated() {
    WorkflowInstance wfInstance = new WorkflowInstance();
    SubworkflowInitiator initiator = new SubworkflowInitiator();
    UpstreamInitiator.Info info = new UpstreamInitiator.Info();
    info.setWorkflowId("foo");
    info.setInstanceId(123L);
    info.setRunId(2L);
    info.setStepId("bar");
    initiator.setAncestors(Collections.singletonList(info));
    wfInstance.setInitiator(initiator);
    wfInstance.setStatus(WorkflowInstance.Status.FAILED);
    wfInstance.setWorkflowInstanceId(10L);
    wfInstance.setWorkflowRunId(1L);
    wfInstance.setWorkflowId("test-workflow");
    wfInstance.setRuntimeWorkflow(Workflow.builder().build());
    RunRequest request =
        RunRequest.builder()
            .initiator(new ManualInitiator())
            .currentPolicy(RunPolicy.RESTART_FROM_BEGINNING)
            .restartConfig(
                RestartConfig.builder().addRestartNode("test-workflow", 1L, null).build())
            .build();

    WorkflowInstance parentInstance = new WorkflowInstance();
    parentInstance.setWorkflowId("foo");
    parentInstance.setWorkflowInstanceId(123L);
    parentInstance.setStatus(WorkflowInstance.Status.IN_PROGRESS);
    parentInstance.setAggregatedInfo(new WorkflowInstanceAggregatedInfo());
    parentInstance
        .getAggregatedInfo()
        .setStepAggregatedViews(
            Collections.singletonMap(
                "bar", StepAggregatedView.builder().status(StepInstance.Status.RUNNING).build()));

    when(instanceDao.getWorkflowInstance("test-workflow", 1L, Constants.LATEST_INSTANCE_RUN, true))
        .thenReturn(wfInstance);
    when(instanceDao.getWorkflowInstance("foo", 123L, Constants.LATEST_INSTANCE_RUN, true))
        .thenReturn(parentInstance);

    RunResponse response = actionHandler.restart(request);
    assertEquals("foo", response.getWorkflowId());
    assertEquals(123L, response.getWorkflowInstanceId());
    assertEquals(RunResponse.Status.DELEGATED, response.getStatus());

    parentInstance
        .getAggregatedInfo()
        .setStepAggregatedViews(
            Collections.singletonMap(
                "bar",
                StepAggregatedView.builder().status(StepInstance.Status.NOT_CREATED).build()));
    AssertHelper.assertThrows(
        "step id is not in the DAG",
        IllegalArgumentException.class,
        "step [bar] is not created in the DAG",
        () -> actionHandler.restart(request));
  }