Object getFromForeach(String foreachStepId, String stepId, String paramName) {
    try {
      return executor
          .submit(() -> fromForeach(foreachStepId, stepId, paramName))
          .get(TIMEOUT_IN_MILLIS, TimeUnit.MILLISECONDS);
    } catch (Exception e) {
      throw new MaestroInternalError(
          e,
          "getFromForeach throws an exception for foreachStepId=[%s], stepId=[%s], paramName=[%s]",
          foreachStepId,
          stepId,
          paramName);
    }
  }
@Test
  public void testInvalidGetFromForeach() throws Exception {
    AssertHelper.assertThrows(
        "Cannot find the referenced step id",
        MaestroInternalError.class,
        "getFromForeach throws an exception",
        () -> paramExtension.getFromForeach("non-existing-job", "job1", "sleep_seconds"));

    StepRuntimeSummary summary = StepRuntimeSummary.builder().type(StepType.NOOP).build();

    when(allStepOutputData.get("foreach-job"))
        .thenReturn(Collections.singletonMap("maestro_step_runtime_summary", summary));
    AssertHelper.assertThrows(
        "Only foreach step can call getFromForeach",
        MaestroInternalError.class,
        "getFromForeach throws an exception",
        () -> paramExtension.getFromForeach("foreach-job", "job1", "sleep_seconds"));

    summary = loadObject(TEST_STEP_RUNTIME_SUMMARY, StepRuntimeSummary.class);
    summary.getArtifacts().remove(Artifact.Type.FOREACH.key());
    when(allStepOutputData.get("foreach-job"))
        .thenReturn(Collections.singletonMap("maestro_step_runtime_summary", summary));
    AssertHelper.assertThrows(
        "Cannot load param from uninitialized foreach",
        MaestroInternalError.class,
        "getFromForeach throws an exception",
        () -> paramExtension.getFromForeach("foreach-job", "job1", "sleep_seconds"));

    summary = loadObject(TEST_STEP_RUNTIME_SUMMARY, StepRuntimeSummary.class);
    when(allStepOutputData.get("foreach-job"))
        .thenReturn(Collections.singletonMap("maestro_step_runtime_summary", summary));
    when(stepInstanceDao.getForeachParamType(any(), any(), any())).thenReturn(ParamType.LONG_ARRAY);
    when(stepInstanceDao.getEvaluatedResultsFromForeach(any(), any(), any()))
        .thenReturn(Collections.singletonMap(1L, "12"));

    AssertHelper.assertThrows(
        "cannot get non-primitive type param from foreach",
        MaestroInternalError.class,
        "getFromForeach throws an exception",
        () -> paramExtension.getFromForeach("foreach-job", "job1", "sleep_seconds"));
  }