@VisibleForTesting
  static Map<String, Long> getStepIdToRunIdForForeachAndSubworkflowFromPreviousRuns(
      WorkflowInstance instance) {
    Map<String, StepType> stepIdToStepTypeForForeachAndSubworkflows =
        instance.getRuntimeWorkflow().getSteps().stream()
            .filter(
                step ->
                    step.getType().equals(StepType.FOREACH)
                        || step.getType().equals(StepType.SUBWORKFLOW))
            .collect(Collectors.toMap(Step::getId, Step::getType));

    if (stepIdToStepTypeForForeachAndSubworkflows.isEmpty()) {
      // if no foreach and subworkflow steps in the workflow definition
      // result should be empty
      return Collections.emptyMap();
    }

    // stepIdToRunId for subworkflow and foreach steps that
    // are not part of current instance's runtimeDAG
    return instance.getAggregatedInfo().getStepAggregatedViews().entrySet().stream()
        .filter(
            step ->
                !instance.getRuntimeDag().containsKey(step.getKey())
                    && stepIdToStepTypeForForeachAndSubworkflows.containsKey(step.getKey()))
        .collect(Collectors.toMap(Map.Entry::getKey, s -> s.getValue().getWorkflowRunId()));
  }
@Test
  public void testGetStepIdToRunIdForForeachAndSubworkflowFromPreviousRuns() throws IOException {
    WorkflowInstance sampleInstance =
        loadObject(
            "fixtures/instances/sample-workflow-instance-created-foreach-subworkflow-1.json",
            WorkflowInstance.class);

    Map<String, Long> stepIdRunId =
        RollupAggregationHelper.getStepIdToRunIdForForeachAndSubworkflowFromPreviousRuns(
            sampleInstance);

    // both subworkflow and foreach steps are included in the current run
    assertEquals(0, stepIdRunId.size());

    sampleInstance.getRuntimeDag().remove("job_subworkflow");

    stepIdRunId =
        RollupAggregationHelper.getStepIdToRunIdForForeachAndSubworkflowFromPreviousRuns(
            sampleInstance);

    // foreach step is included in the current run
    // while subworkflow one is not
    assertEquals(1, stepIdRunId.size());
    assertEquals(3L, stepIdRunId.get("job_subworkflow").longValue());

    sampleInstance.getRuntimeDag().remove("job_foreach");

    stepIdRunId =
        RollupAggregationHelper.getStepIdToRunIdForForeachAndSubworkflowFromPreviousRuns(
            sampleInstance);

    // both are included
    assertEquals(2, stepIdRunId.size());
    assertEquals(3L, stepIdRunId.get("job_subworkflow").longValue());
    assertEquals(1L, stepIdRunId.get("job_foreach").longValue());

    WorkflowInstance sampleInstanceNoForeachSubworkflow =
        loadObject(
            "fixtures/instances/sample-workflow-instance-created.json", WorkflowInstance.class);

    stepIdRunId =
        RollupAggregationHelper.getStepIdToRunIdForForeachAndSubworkflowFromPreviousRuns(
            sampleInstanceNoForeachSubworkflow);

    assertEquals(0, stepIdRunId.size());
  }