@VisibleForTesting
  @Override
  List<String> cancelNonTerminalTasks(Workflow workflow) {
    List<String> erroredTasks = new ArrayList<>();
    // Update non-terminal tasks' status to CANCELED
    for (Task task : workflow.getTasks()) {
      if (!task.getStatus().isTerminal()) {
        // Cancel the ones which are not completed yet....
        task.setStatus(CANCELED);
        // all of our tasks are system tasks.
        Checks.checkTrue(
            SystemTaskType.is(task.getTaskType()),
            "Invalid task type [%s], all tasks should have a known maestro task type.",
            task.getTaskType());
        WorkflowSystemTask workflowSystemTask = WorkflowSystemTask.get(task.getTaskType());
        try {
          workflowSystemTask.cancel(workflow, task, this);
          executionDAOFacade.updateTask(task); // only update if cancelled
        } catch (Exception e) {
          erroredTasks.add(task.getReferenceTaskName());
          LOG.error(
              "Error canceling system task:{}/{} in workflow: {}",
              workflowSystemTask.getName(),
              task.getTaskId(),
              workflow.getWorkflowId(),
              e);
        }
      }
    }
    if (erroredTasks.isEmpty()) {
      try {
        workflowStatusListener.onWorkflowFinalizedIfEnabled(workflow);
        queueDAO.remove(DECIDER_QUEUE, workflow.getWorkflowId());
      } catch (Exception e) {
        LOG.error("Error removing workflow: {} from decider queue", workflow.getWorkflowId(), e);
        throw e; // we need to throw it to get at least once guarantee.
      }
    } else {
      // also throw to retry errored tasks later.
      throw new MaestroRetryableError(
          "Error canceling tasks [%s] in workflow: [%s]", erroredTasks, workflow.getWorkflowId());
    }
    return erroredTasks;
  }
@Test
  public void testCancelNonTerminalTasksFailed() {
    Task mockTask1 = mock(Task.class);
    when(mockTask1.getTaskId()).thenReturn("task-id-1");
    when(mockTask1.getTaskType()).thenReturn(Constants.MAESTRO_PREFIX);
    when(mockTask1.getStatus()).thenReturn(Task.Status.IN_PROGRESS);

    workflow.getTasks().add(mockTask1);

    AssertHelper.assertThrows(
        "All tasks should have a known maestro task type.",
        IllegalArgumentException.class,
        "Invalid task type",
        () -> maestroWorkflowExecutor.cancelNonTerminalTasks(workflow));

    when(mockTask1.getTaskType()).thenReturn(Constants.DEFAULT_START_STEP_NAME);
    task1.setShouldThrow(true);
    Task maestroTask = new Task();
    maestroTask.setTaskId(UUID.randomUUID().toString());
    maestroTask.setTaskType(Constants.MAESTRO_TASK_NAME);
    maestroTask.setStatus(Task.Status.SCHEDULED);

    workflow.getTasks().add(maestroTask);

    AssertHelper.assertThrows(
        "Cancel throws an exception will fail cancelNonTerminalTasks call.",
        MaestroRetryableError.class,
        "Error canceling tasks ",
        () -> maestroWorkflowExecutor.cancelNonTerminalTasks(workflow));
    ArgumentCaptor<Task> argumentCaptor = ArgumentCaptor.forClass(Task.class);
    verify(executionDAOFacade, times(1)).updateTask(argumentCaptor.capture());
    assertEquals(1, argumentCaptor.getAllValues().size());
    assertEquals(Constants.MAESTRO_TASK_NAME, argumentCaptor.getAllValues().get(0).getTaskType());
    assertEquals(Task.Status.CANCELED, argumentCaptor.getAllValues().get(0).getStatus());
    verify(queueDAO, times(0)).remove(any(), any());
  }