public Optional<Details> runForeachBatch(
      Workflow workflow,
      Long internalId,
      long workflowVersionId,
      RunProperties runProperties,
      String foreachStepId,
      ForeachArtifact artifact,
      List<RunRequest> requests,
      List<Long> instanceIds,
      int batchSize) {
    if (ObjectHelper.isCollectionEmptyOrNull(requests)) {
      return Optional.empty();
    }
    Checks.checkTrue(
        requests.size() == instanceIds.size(),
        "Run request list size [%s] must match instance id list size [%s]",
        requests.size(),
        instanceIds.size());
    List<WorkflowInstance> instances;
    if (artifact.isFreshRun()) {
      instances =
          createStartForeachInstances(
              workflow,
              internalId,
              workflowVersionId,
              artifact.getForeachRunId(),
              runProperties,
              requests,
              instanceIds);
    } else {
      instances =
          createRestartForeachInstances(
              workflow,
              internalId,
              workflowVersionId,
              runProperties,
              foreachStepId,
              artifact,
              requests,
              instanceIds);
    }
    if (ObjectHelper.isCollectionEmptyOrNull(instances)) {
      return Optional.empty();
    }
    return instanceDao.runWorkflowInstances(workflow.getId(), instances, batchSize);
  }
@Test
  public void testCreateRestartForeachInstancesUpstreamModeFromIncomplete() {
    doNothing().when(workflowHelper).updateWorkflowInstance(any(), any());
    when(instanceDao.getLatestWorkflowInstanceRun(anyString(), anyLong()))
        .thenReturn(new WorkflowInstance());

    Map<String, Map<String, ParamDefinition>> stepRunParams =
        Collections.singletonMap(
            "job1",
            Collections.singletonMap("p1", ParamDefinition.buildParamDefinition("p1", "d1")));

    ForeachArtifact artifact = new ForeachArtifact();
    artifact.setRunPolicy(RunPolicy.RESTART_FROM_INCOMPLETE);
    artifact.setTotalLoopCount(10);
    artifact.setForeachWorkflowId("maestro_foreach_x");
    artifact.setAncestorIterationCount(3L);
    artifact.setForeachRunId(3L);
    artifact.setForeachOverview(new ForeachStepOverview());
    artifact.getForeachOverview().addOne(2, WorkflowInstance.Status.FAILED, null);
    RestartConfig restartConfig =
        RestartConfig.builder()
            .restartPolicy(RunPolicy.RESTART_FROM_INCOMPLETE)
            .downstreamPolicy(RunPolicy.RESTART_FROM_INCOMPLETE)
            .stepRestartParams(stepRunParams)
            .addRestartNode("maestro_foreach_x", 1, null)
            .build();

    ForeachInitiator initiator = new ForeachInitiator();
    UpstreamInitiator.Info parent = new UpstreamInitiator.Info();
    parent.setWorkflowId("maestro_foreach_x");
    parent.setInstanceId(1);
    parent.setRunId(1);
    parent.setStepId("foreach-step");
    parent.setStepAttemptId(1);
    initiator.setAncestors(Collections.singletonList(parent));

    RunRequest runRequest =
        RunRequest.builder()
            .initiator(initiator)
            .currentPolicy(RunPolicy.RESTART_FROM_INCOMPLETE)
            .restartConfig(restartConfig)
            .build();

    Optional<Details> errors =
        actionHandler.runForeachBatch(
            definition.getWorkflow(),
            123L,
            10L,
            new RunProperties(),
            "foreach-step",
            artifact,
            Collections.singletonList(runRequest),
            Collections.singletonList(2L),
            3);
    assertFalse(errors.isPresent());

    verify(instanceDao, times(1)).runWorkflowInstances(any(), any(), anyInt());

    ArgumentCaptor<RunRequest> captor = ArgumentCaptor.forClass(RunRequest.class);
    verify(workflowHelper, times(1)).updateWorkflowInstance(any(), captor.capture());

    RunRequest res = captor.getValue();
    assertEquals(RunPolicy.RESTART_FROM_INCOMPLETE, res.getCurrentPolicy());
    // it will keep the restart config with step restart params
    assertEquals(restartConfig, res.getRestartConfig());
  }