public RunResponse restartDirectly(WorkflowInstance instance, RunRequest runRequest) {
    Checks.checkTrue(
        !runRequest.isFreshRun(),
        "Cannot restart a workflow instance %s using fresh run policy [%s]",
        instance.getIdentity(),
        runRequest.getCurrentPolicy());
    if (runRequest.getRestartConfig() != null) {
      runRequest.validateIdentity(instance);
    }

    WorkflowInstance.Status status =
        instanceDao.getLatestWorkflowInstanceStatus(
            instance.getWorkflowId(), instance.getWorkflowInstanceId());
    if (!status.isTerminal()) {
      throw new MaestroInvalidStatusException(
          "Cannot restart workflow instance [%s][%s] as the latest run status is [%s] (non-terminal).",
          instance.getWorkflowId(), instance.getWorkflowInstanceId(), status);
    }

    workflowHelper.updateWorkflowInstance(instance, runRequest);
    int ret =
        runStrategyDao.startWithRunStrategy(
            instance, workflowDao.getRunStrategy(instance.getWorkflowId()));
    RunResponse runResponse = RunResponse.from(instance, ret);
    LOG.info("Restart a workflow instance with a response: [{}]", runResponse);
    return runResponse;
  }
@Test
  public void testRestartSubworkflow() {
    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.SUCCEEDED);
    wfInstance.setWorkflowInstanceId(10L);
    wfInstance.setWorkflowRunId(3L);
    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", 10L, null).build())
            .build();
    when(runStrategyDao.startWithRunStrategy(any(), any())).thenReturn(1);
    when(instanceDao.getLatestWorkflowInstanceStatus(any(), anyLong()))
        .thenReturn(WorkflowInstance.Status.SUCCEEDED);
    doNothing().when(workflowHelper).updateWorkflowInstance(any(), any());

    RunResponse response = actionHandler.restartDirectly(wfInstance, request);
    assertEquals("test-workflow", response.getWorkflowId());
    assertEquals(10L, response.getWorkflowInstanceId());
    assertEquals(3L, response.getWorkflowRunId());
    assertEquals(RunResponse.Status.WORKFLOW_RUN_CREATED, response.getStatus());
    verify(workflowHelper, times(1)).updateWorkflowInstance(any(), any());
  }