public static Set<String> computeStepIdsInRuntimeDag(
      @NotNull WorkflowInstance instance, @NotNull Set<String> knownStepIds) {
    if (instance.isFreshRun()
        || instance.getRunConfig() == null
        || instance.getRunConfig().getPolicy() != RunPolicy.RESTART_FROM_SPECIFIC) {
      return instance.getRuntimeDag().keySet();
    }

    Map<String, Set<String>> childMap = new HashMap<>();
    prepareDagForTraversal(
        instance.getRuntimeDag(),
        knownStepIds,
        instance.getRunConfig().getRestartConfig(),
        new HashMap<>(),
        childMap);

    return childMap.keySet();
  }
@Test
  public void testComputeStepIdsInRuntimeDagForStepRestart() {
    WorkflowInstance instance = new WorkflowInstance();
    instance.setRunConfig(new RunConfig());
    instance.getRunConfig().setPolicy(RunPolicy.RESTART_FROM_SPECIFIC);
    instance
        .getRunConfig()
        .setRestartConfig(
            RestartConfig.builder().addRestartNode("sample-dag-test-1", 1, "job_9").build());
    instance.setRuntimeDag(runtimeDag1);
    Set<String> actual =
        DagHelper.computeStepIdsInRuntimeDag(instance, Collections.singleton("job_9"));
    Assert.assertEquals("[job_9, job_8]", actual.toString());
  }