@Override
  public Map<String, StepTransition> translate(WorkflowInstance workflowInstance) {
    WorkflowInstance instance = objectMapper.convertValue(workflowInstance, WorkflowInstance.class);

    if (instance.getRunConfig() != null) {
      if (instance.getRunConfig().getPolicy() == RunPolicy.RESTART_FROM_INCOMPLETE
          || instance.getRunConfig().getPolicy() == RunPolicy.RESTART_FROM_SPECIFIC) {
        Map<String, StepInstance.Status> statusMap =
            instance.getAggregatedInfo().getStepAggregatedViews().entrySet().stream()
                .collect(
                    Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().getStatus()));
        if (!statusMap.isEmpty()) {
          instance
              .getRunConfig()
              .setStartStepIds(
                  statusMap.entrySet().stream()
                      .filter(
                          entry ->
                              !entry.getValue().isComplete()
                                  && (entry.getValue().isTerminal()
                                      || entry.getValue() == StepInstance.Status.NOT_CREATED))
                      .map(Map.Entry::getKey)
                      .collect(Collectors.toList()));
        }
        // handle the special case of restarting from a completed step
        if (instance.getRunConfig().getPolicy() == RunPolicy.RESTART_FROM_SPECIFIC) {
          String restartStepId =
              RunRequest.getCurrentNode(instance.getRunConfig().getRestartConfig()).getStepId();
          if (!instance.getRunConfig().getStartStepIds().contains(restartStepId)) {
            instance.getRunConfig().getStartStepIds().add(restartStepId);
          }
        }
      } else {
        if (workflowInstance.getRunConfig().getStartStepIds() != null) {
          instance
              .getRunConfig()
              .setStartStepIds(new ArrayList<>(workflowInstance.getRunConfig().getStartStepIds()));
        }
        if (workflowInstance.getRunConfig().getEndStepIds() != null) {
          instance
              .getRunConfig()
              .setEndStepIds(new ArrayList<>(workflowInstance.getRunConfig().getEndStepIds()));
        }
      }
    }

    List<String> startStepIds =
        instance.getRunConfig() != null && instance.getRunConfig().getStartStepIds() != null
            ? instance.getRunConfig().getStartStepIds()
            : null;
    List<String> endStepIds =
        instance.getRunConfig() != null && instance.getRunConfig().getEndStepIds() != null
            ? instance.getRunConfig().getEndStepIds()
            : null;

    return WorkflowGraph.computeDag(instance.getRuntimeWorkflow(), startStepIds, endStepIds);
  }
@Test
  public void testTranslateForRestartFromSpecificWithCompleteBranch() {
    instance.getRuntimeWorkflow().getSteps().get(2).getTransition().getSuccessors().remove("job.2");
    instance
        .getAggregatedInfo()
        .getStepAggregatedViews()
        .put("job.2", StepAggregatedView.builder().status(StepInstance.Status.STOPPED).build());
    instance.getRunConfig().setPolicy(RunPolicy.RESTART_FROM_SPECIFIC);
    instance
        .getRunConfig()
        .setRestartConfig(
            RestartConfig.builder()
                .addRestartNode("sample-dag-test-3", 1, "job1")
                .restartPolicy(RunPolicy.RESTART_FROM_SPECIFIC)
                .build());

    Map<String, StepTransition> dag = translator.translate(instance);
    Assert.assertEquals(
        new HashSet<>(Arrays.asList("job1", "job.2", "job3", "job4")), dag.keySet());
  }