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 testStop() {
    StepInstanceActionResponse response = actionDao.terminate(instance, "job1", user, STOP, false);
    Assert.assertEquals("sample-dag-test-3", response.getWorkflowId());
    Assert.assertEquals(1, response.getWorkflowInstanceId());
    Assert.assertEquals(1, response.getWorkflowRunId());
    Assert.assertEquals("job1", response.getStepId());
    Assert.assertEquals(1L, response.getStepAttemptId().longValue());
    Assert.assertEquals(
        "User [tester] take action [STOP] on the step due to reason: [manual step instance API call]",
        response.getTimelineEvent().getMessage());
    Mockito.verify(publisher, Mockito.times(1)).publish(any(StepInstanceWakeUpEvent.class));
  }