public Optional<Details> runWorkflowInstances(
      String workflowId, List<WorkflowInstance> instances, int batchSize) {
    Checks.checkTrue(
        !ObjectHelper.isCollectionEmptyOrNull(instances),
        "cannot run null or empty workflow instances for %s",
        workflowId);
    updateInstances(instances);
    try {
      int[] res =
          withRetryableTransaction(
              conn -> {
                tryUpdateAncestorRunsStatus(
                    conn,
                    workflowId,
                    instances.get(0).getWorkflowInstanceId(),
                    instances.get(instances.size() - 1));
                return insertMaestroWorkflowInstances(conn, instances);
              });
      LOG.debug(
          "Created workflow instances {} for workflow_id [{}]", Arrays.toString(res), workflowId);
      int cnt = publishRunInstancesJobEvents(workflowId, instances, batchSize);
      LOG.info(
          "Created {}/{} workflow instances and sent {} run job events for workflow id {}",
          res.length,
          instances.size(),
          cnt,
          workflowId);
      return Optional.empty();
    } catch (MaestroInternalError error) { // non-retryable error
      throw error;
    } catch (MaestroRetryableError retryableError) {
      return Optional.of(retryableError.getDetails());
    } catch (RuntimeException e) {
      LOG.warn(
          "Failed to create workflow instance batch (starting at {}) for workflow [{}] due to",
          instances.get(0).getWorkflowInstanceId(),
          workflowId,
          e);
      return Optional.of(
          Details.create(e, true, "ERROR: failed creating workflow instance batch with an error"));
    }
  }
@Test
  public void testRunWorkflowInstances() throws Exception {
    WorkflowInstance wfi1 = loadObject(TEST_WORKFLOW_INSTANCE, WorkflowInstance.class);
    wfi1.setWorkflowUuid("wfi1-uuid");
    wfi1.setWorkflowInstanceId(100L);
    WorkflowInstance wfi2 = loadObject(TEST_WORKFLOW_INSTANCE, WorkflowInstance.class);
    wfi2.setWorkflowUuid("wfi2-uuid");
    wfi2.setWorkflowInstanceId(101L);
    WorkflowInstance wfi3 = loadObject(TEST_WORKFLOW_INSTANCE, WorkflowInstance.class);
    wfi3.setWorkflowUuid("wfi3-uuid");
    wfi3.setWorkflowInstanceId(102L);

    Optional<Details> res =
        instanceDao.runWorkflowInstances(TEST_WORKFLOW_ID, Arrays.asList(wfi1, wfi2, wfi3), 2);
    assertFalse(res.isPresent());
    assertEquals(100, wfi1.getWorkflowInstanceId());
    assertEquals(1, wfi1.getWorkflowRunId());
    assertEquals(WorkflowInstance.Status.CREATED, wfi1.getStatus());
    assertEquals(101, wfi2.getWorkflowInstanceId());
    assertEquals(1, wfi2.getWorkflowRunId());
    assertEquals(WorkflowInstance.Status.CREATED, wfi2.getStatus());
    assertEquals(102, wfi3.getWorkflowInstanceId());
    assertEquals(1, wfi3.getWorkflowRunId());
    assertEquals(WorkflowInstance.Status.CREATED, wfi3.getStatus());
    verify(publisher, times(2)).publishOrThrow(any(), any());

    WorkflowInstance actual =
        instanceDao.getLatestWorkflowInstanceRun(TEST_WORKFLOW_ID, wfi3.getWorkflowInstanceId());
    assertEquals(102, actual.getWorkflowInstanceId());
    assertEquals(1, actual.getWorkflowRunId());
    assertEquals(WorkflowInstance.Status.CREATED, actual.getStatus());

    reset(publisher);
    wfi3.setWorkflowUuid("wfi-uuid");
    wfi3.setWorkflowInstanceId(103L);

    doAnswer(
            (Answer<Void>)
                invocation -> {
                  RunWorkflowInstancesJobEvent jobEvent =
                      (RunWorkflowInstancesJobEvent) invocation.getArguments()[0];
                  assertEquals(TEST_WORKFLOW_ID, jobEvent.getWorkflowId());
                  assertEquals(3, jobEvent.size());
                  assertEquals(103, jobEvent.getInstanceRunUuids().get(2).getInstanceId());
                  assertEquals("wfi-uuid", jobEvent.getInstanceRunUuids().get(2).getUuid());
                  return null;
                })
        .when(publisher)
        .publishOrThrow(any(), any());

    res = instanceDao.runWorkflowInstances(TEST_WORKFLOW_ID, Arrays.asList(wfi1, wfi2, wfi3), 5);
    assertFalse(res.isPresent());
    actual =
        instanceDao.getLatestWorkflowInstanceRun(TEST_WORKFLOW_ID, wfi3.getWorkflowInstanceId());
    assertEquals(103, actual.getWorkflowInstanceId());
    assertEquals(1, actual.getWorkflowRunId());
    assertEquals(WorkflowInstance.Status.CREATED, actual.getStatus());

    MaestroTestHelper.removeWorkflowInstance(dataSource, TEST_WORKFLOW_ID, 100);
    MaestroTestHelper.removeWorkflowInstance(dataSource, TEST_WORKFLOW_ID, 101);
    MaestroTestHelper.removeWorkflowInstance(dataSource, TEST_WORKFLOW_ID, 102);
    MaestroTestHelper.removeWorkflowInstance(dataSource, TEST_WORKFLOW_ID, 103);
  }