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 testRestartRunForeachBatch() {
    ForeachArtifact artifact = new ForeachArtifact();
    artifact.setRunPolicy(RunPolicy.RESTART_FROM_INCOMPLETE);
    artifact.setTotalLoopCount(10);
    artifact.setForeachWorkflowId(instance.getWorkflowId());
    artifact.setForeachRunId(3L);
    RunRequest request =
        RunRequest.builder()
            .initiator(new ManualInitiator())
            .currentPolicy(RunPolicy.START_FRESH_NEW_RUN)
            .build();
    Optional<Details> errors =
        actionHandler.runForeachBatch(
            definition.getWorkflow(),
            123L,
            10L,
            new RunProperties(),
            "foreach-step",
            artifact,
            Collections.singletonList(request),
            Collections.singletonList(5L),
            1);
    assertFalse(errors.isPresent());
    ArgumentCaptor<List<WorkflowInstance>> captor = ArgumentCaptor.forClass(List.class);
    verify(instanceDao, times(1))
        .runWorkflowInstances(eq(artifact.getForeachWorkflowId()), captor.capture(), eq(1));
    List<WorkflowInstance> res = captor.getValue();
    assertEquals(1, res.size());
    assertEquals(artifact.getForeachWorkflowId(), res.get(0).getWorkflowId());
    assertEquals(5L, res.get(0).getWorkflowInstanceId());
    assertEquals(3L, res.get(0).getWorkflowRunId());
    assertEquals(10L, res.get(0).getWorkflowVersionId());
  }