public int terminateQueuedInstances(
      String workflowId, int limit, WorkflowInstance.Status status, String reason) {
    TimelineEvent timelineEvent =
        TimelineLogEvent.warn(TERMINATION_MESSAGE_TEMPLATE, status.name(), reason);
    String timelineEventStr = toJson(timelineEvent);
    return withMetricLogError(
        () ->
            withRetryableTransaction(
                conn -> {
                  List<WorkflowInstance> stoppedInstances = new ArrayList<>();
                  try (PreparedStatement stmt =
                      conn.prepareStatement(TERMINATE_QUEUED_INSTANCES_QUERY)) {
                    int idx = 0;
                    stmt.setString(++idx, status.name());
                    stmt.setString(++idx, timelineEventStr);
                    stmt.setString(++idx, workflowId);
                    stmt.setInt(++idx, limit);
                    try (ResultSet result = stmt.executeQuery()) {
                      while (result.next()) {
                        WorkflowInstance instance =
                            fromJson(result.getString(1), WorkflowInstance.class);
                        stoppedInstances.add(instance);
                      }
                    }
                    if (!stoppedInstances.isEmpty()) {
                      WorkflowInstanceUpdateJobEvent jobEvent =
                          WorkflowInstanceUpdateJobEvent.create(
                              stoppedInstances, status, System.currentTimeMillis());
                      publisher.publishOrThrow(
                          jobEvent,
                          "Failed sending an update job event to notify stopping workflow instances.");
                    }
                  }
                  return stoppedInstances.size();
                }),
        "terminateQueuedInstances",
        "Failed to terminate the queued workflow instances for workflow {}",
        workflowId);
  }
@Test
  public void testTerminateQueuedInstances() 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());
    int cnt =
        instanceDao.terminateQueuedInstances(
            TEST_WORKFLOW_ID, 3, WorkflowInstance.Status.STOPPED, "test-reason");
    assertEquals(3, cnt);
    verify(publisher, times(1)).publishOrThrow(any(WorkflowInstanceUpdateJobEvent.class), any());
    cnt =
        instanceDao.terminateQueuedInstances(
            TEST_WORKFLOW_ID, 2, WorkflowInstance.Status.FAILED, "test-reason");
    assertEquals(1, cnt);
    verify(publisher, times(2)).publishOrThrow(any(WorkflowInstanceUpdateJobEvent.class), any());
    cnt =
        instanceDao.terminateQueuedInstances(
            TEST_WORKFLOW_ID, 1, WorkflowInstance.Status.STOPPED, "test-reason");
    assertEquals(0, cnt);
    verify(publisher, times(2)).publishOrThrow(any(WorkflowInstanceUpdateJobEvent.class), any());
    MaestroTestHelper.removeWorkflowInstance(dataSource, TEST_WORKFLOW_ID, 100);
    MaestroTestHelper.removeWorkflowInstance(dataSource, TEST_WORKFLOW_ID, 101);
    MaestroTestHelper.removeWorkflowInstance(dataSource, TEST_WORKFLOW_ID, 102);
  }