public static EnumMap<StepInstance.Status, WorkflowStepStatusSummary> toStepStatusMap(
      WorkflowSummary summary, Map<String, StepRuntimeState> states) {
    AtomicLong ordinal = new AtomicLong(0);
    Map<String, Long> stepOrdinalMap =
        summary.getRuntimeDag().keySet().stream()
            .collect(Collectors.toMap(Function.identity(), s -> ordinal.incrementAndGet()));

    EnumMap<StepInstance.Status, WorkflowStepStatusSummary> stats =
        new EnumMap<>(StepInstance.Status.class);
    states.forEach(
        (stepId, state) -> {
          if (state.getStatus() != StepInstance.Status.NOT_CREATED) {
            if (!stats.containsKey(state.getStatus())) {
              stats.put(state.getStatus(), new WorkflowStepStatusSummary());
            }
            List<Long> stepInfo =
                Arrays.asList(
                    stepOrdinalMap.remove(stepId), state.getStartTime(), state.getEndTime());
            stats.get(state.getStatus()).addStep(stepInfo);
          }
        });

    stats.forEach((status, stepStatusSummary) -> stepStatusSummary.sortSteps());
    // Don't include NOT_CREATED in the stats. Use decode method to get them if needed.
    return stats;
  }
@Test
  public void testToStepStatusMap() throws Exception {
    WorkflowSummary workflowSummary =
        loadObject("fixtures/parameters/sample-wf-summary-params.json", WorkflowSummary.class);
    StepRuntimeState state = new StepRuntimeState();
    state.setStatus(StepInstance.Status.RUNNING);
    state.setStartTime(123L);
    Assert.assertEquals(
        singletonEnumMap(
            StepInstance.Status.RUNNING,
            WorkflowStepStatusSummary.of(0).addStep(Arrays.asList(2L, 123L, null))),
        TaskHelper.toStepStatusMap(workflowSummary, singletonMap("job1", state)));
  }