public StepInstanceActionResponse skip(
      String workflowId,
      long workflowInstanceId,
      String stepId,
      User user,
      RunRequest runRequest,
      boolean blocking) {
    WorkflowInstance instance =
        instanceDao.getWorkflowInstance(
            workflowId, workflowInstanceId, Constants.LATEST_INSTANCE_RUN, true);
    StepAggregatedView view =
        Checks.notNull(
            instance.getAggregatedInfo().getStepAggregatedViews().get(stepId),
            "Invalid: cannot find the step view of workflow step [%s][%s][%s]",
            workflowId,
            workflowInstanceId,
            stepId);
    if (view.getStatus() == StepInstance.Status.NOT_CREATED) {
      throw new MaestroBadRequestException(
          Collections.emptyList(),
          "Cannot skip step [%s][%s][%s] before it is created. Please try it again.",
          workflowId,
          workflowInstanceId,
          stepId);
    }

    List<Actions.StepInstanceAction> supportedActions =
        Actions.STEP_INSTANCE_STATUS_TO_ACTION_MAP.get(view.getStatus());
    if (supportedActions != null && !supportedActions.contains(Actions.StepInstanceAction.SKIP)) {
      throw new MaestroBadRequestException(
          Collections.emptyList(),
          "Cannot skip step [%s][%s][%s] because it is unsupported by the step action map %s",
          workflowId,
          workflowInstanceId,
          stepId,
          supportedActions);
    }

    if (!instance.getStatus().isTerminal() && view.getStatus().shouldWakeup()) {
      return actionDao.terminate(instance, stepId, user, Actions.StepInstanceAction.SKIP, blocking);
    }

    RunResponse runResponse = restart(runRequest, blocking);
    return runResponse.toStepInstanceActionResponse();
  }
@Test
  public void testInvalidSkip() {
    when(instanceDao.getWorkflowInstance(
            "sample-minimal-wf", 1, Constants.LATEST_INSTANCE_RUN, true))
        .thenReturn(instance);
    WorkflowInstanceAggregatedInfo aggregatedInfo = mock(WorkflowInstanceAggregatedInfo.class);
    when(instance.getAggregatedInfo()).thenReturn(aggregatedInfo);
    StepAggregatedView aggregatedView = mock(StepAggregatedView.class);
    when(aggregatedInfo.getStepAggregatedViews()).thenReturn(singletonMap("job1", aggregatedView));
    when(instance.getStatus()).thenReturn(WorkflowInstance.Status.IN_PROGRESS);
    when(aggregatedView.getStatus()).thenReturn(StepInstance.Status.RUNNING);
    AssertHelper.assertThrows(
        "Cannot find status in aggregated step views",
        NullPointerException.class,
        "Invalid: cannot find the step view of workflow step ",
        () -> stepActionHandler.skip("sample-minimal-wf", 1, "job2", user, null, true));

    when(aggregatedView.getStatus()).thenReturn(StepInstance.Status.NOT_CREATED);
    AssertHelper.assertThrows(
        "Cannot skip not-created step",
        MaestroBadRequestException.class,
        "Cannot skip step [sample-minimal-wf][1][job1] before it is created. Please try it again.",
        () -> stepActionHandler.skip("sample-minimal-wf", 1, "job1", user, null, true));

    when(aggregatedView.getStatus()).thenReturn(StepInstance.Status.CREATED);
    AssertHelper.assertThrows(
        "Cannot skip not-created step",
        MaestroBadRequestException.class,
        "Cannot skip step [sample-minimal-wf][1][job1] because it is unsupported by the step action map",
        () -> stepActionHandler.skip("sample-minimal-wf", 1, "job1", user, null, true));
  }