public StepInstanceActionResponse terminate(
      WorkflowInstance instance,
      String stepId,
      User user,
      Actions.StepInstanceAction action,
      boolean blocking) {
    validateStepId(instance, stepId, action);

    StepInstance stepInstance =
        stepInstanceDao.getStepInstance(
            instance.getWorkflowId(),
            instance.getWorkflowInstanceId(),
            instance.getWorkflowRunId(),
            stepId,
            Constants.LATEST_INSTANCE_RUN);

    if (!stepInstance.getRuntimeState().getStatus().shouldWakeup()) {
      throw new MaestroInvalidStatusException(
          "Cannot manually %s the step %s as it is in a terminal state [%s]",
          action.name(), stepInstance.getIdentity(), stepInstance.getRuntimeState().getStatus());
    }

    // prepare payload and then add it to db
    StepAction stepAction =
        StepAction.createTerminate(
            action, stepInstance, user, "manual step instance API call", false);
    saveAction(stepInstance, stepAction);

    if (blocking) {
      long startTime = System.currentTimeMillis();
      while (System.currentTimeMillis() - startTime < ACTION_TIMEOUT) {
        StepRuntimeState state =
            stepInstanceDao.getStepInstanceRuntimeState(
                stepInstance.getWorkflowId(),
                stepInstance.getWorkflowInstanceId(),
                stepInstance.getWorkflowRunId(),
                stepInstance.getStepId(),
                Constants.LATEST_INSTANCE_RUN);
        if (!state.getStatus().shouldWakeup()) {
          return createActionResponseFrom(stepInstance, state, stepAction.toTimelineEvent());
        }
        TimeUtils.sleep(CHECK_INTERVAL);
      }

      throw new MaestroTimeoutException(
          "%s action for the step %s is timed out. No retry is needed and maestro will eventually complete the action.",
          action.name(), stepInstance.getIdentity());
    } else {
      return createActionResponseFrom(stepInstance, null, stepAction.toTimelineEvent());
    }
  }
@Test
  public void testInvalidTerminate() {
    AssertHelper.assertThrows(
        "Cannot manually terminate the step",
        MaestroBadRequestException.class,
        "Cannot manually STOP the step [not-existing] because the latest workflow run",
        () -> actionDao.terminate(instance, "not-existing", user, STOP, false));

    AssertHelper.assertThrows(
        "Cannot manually terminate the step",
        MaestroNotFoundException.class,
        "step instance [job.2][LATEST] not found (either not created or deleted)",
        () -> actionDao.terminate(instance, "job.2", user, KILL, false));

    stepInstance.getRuntimeState().setStatus(StepInstance.Status.FATALLY_FAILED);
    stepInstanceDao.insertOrUpsertStepInstance(stepInstance, true);
    AssertHelper.assertThrows(
        "Cannot manually restart the step",
        MaestroInvalidStatusException.class,
        "Cannot manually SKIP the step [sample-dag-test-3][1][1][job1] as it is in a terminal state [FATALLY_FAILED]",
        () -> actionDao.terminate(instance, "job1", user, SKIP, false));

    stepInstance.getRuntimeState().setStatus(StepInstance.Status.RUNNING);
    stepInstanceDao.insertOrUpsertStepInstance(stepInstance, true);
    actionDao.terminate(instance, "job1", user, KILL, false);

    AssertHelper.assertThrows(
        "Cannot manually terminate the step",
        MaestroResourceConflictException.class,
        "There is an ongoing action for this step [sample-dag-test-3][1][1][job1]",
        () -> actionDao.terminate(instance, "job1", user, STOP, false));
  }