@Override
  public void executeSystemTask(WorkflowSystemTask systemTask, String taskId, int callbackTime) {
    try {
      Task task = executionDAOFacade.getTaskById(taskId);
      if (task == null) {
        LOG.error("TaskId: {} could not be found while executing SystemTask", taskId);
        return;
      }
      LOG.debug("Task: {} fetched from execution DAO for taskId: {}", task, taskId);
      String queueName = QueueUtils.getQueueName(task);
      if (task.getStatus().isTerminal()) {
        // Tune the SystemTaskWorkerCoordinator's queues - if the queue size is very big this can
        // happen!
        LOG.info("Task {}/{} was already completed.", task.getTaskType(), task.getTaskId());
        queueDAO.remove(queueName, task.getTaskId());
        return;
      }

      String workflowId = task.getWorkflowInstanceId();
      Workflow workflow = executionDAOFacade.getWorkflowById(workflowId, true);

      if (task.getStartTime() == 0) {
        task.setStartTime(System.currentTimeMillis());
        executionDAOFacade.updateTask(task);
        Monitors.recordQueueWaitTime(task.getTaskDefName(), task.getQueueWaitTime());
      }

      if (workflow.getStatus().isTerminal()) {
        LOG.info(
            "Workflow {} has been completed for {}/{}",
            workflow.getWorkflowId(),
            systemTask.getName(),
            task.getTaskId());
        if (!task.getStatus().isTerminal()) {
          task.setStatus(CANCELED);
        }
        executionDAOFacade.updateTask(task);
        queueDAO.remove(queueName, task.getTaskId());
        return;
      }

      LOG.debug("Executing {}/{}-{}", task.getTaskType(), task.getTaskId(), task.getStatus());
      if (task.getStatus() == SCHEDULED || !systemTask.isAsyncComplete(task)) {
        task.setPollCount(task.getPollCount() + 1);
        // removed poll count DB update here
      }

      deciderService.populateTaskData(task);

      // Stop polling for asyncComplete system tasks that are not in SCHEDULED state
      if (systemTask.isAsyncComplete(task) && task.getStatus() != SCHEDULED) {
        queueDAO.remove(QueueUtils.getQueueName(task), task.getTaskId());
        return;
      }

      taskRunner.runMaestroTask(this, workflow, task, systemTask);

      if (!task.getStatus().isTerminal()) {
        task.setCallbackAfterSeconds(callbackTime);
        try {
          configureCallbackInterval(task); // overwrite if needed
        } catch (Exception e) {
          LOG.error(
              "Error configuring callback interval for task [{}]. Please investigate it",
              task.getTaskId(),
              e);
        }
      }

      updateTask(new TaskResult(task));
      LOG.debug(
          "Done Executing {}/{}-{} output={}",
          task.getTaskType(),
          task.getTaskId(),
          task.getStatus(),
          task.getOutputData());

    } catch (Exception e) {
      Monitors.error("MaestroWorkflowExecutor", "executeSystemTask");
      LOG.error("Error executing system task - {}, with id: {}", systemTask, taskId, e);
    }
  }
@Test
  public void testExecuteSystemTaskPersistStartTime() {
    String workflowId = "workflow-id";
    String taskId = "task-id-1";

    Task maestroTask = new Task();
    maestroTask.setTaskType(Constants.MAESTRO_TASK_NAME);
    maestroTask.setReferenceTaskName("maestroTask");
    maestroTask.setWorkflowInstanceId(workflowId);
    maestroTask.setScheduledTime(System.currentTimeMillis());
    maestroTask.setTaskId(taskId);
    maestroTask.setStatus(Task.Status.IN_PROGRESS);
    maestroTask.setStartTime(0);
    maestroTask.setCallbackAfterSeconds(0);

    Workflow workflow = new Workflow();
    workflow.setWorkflowId(workflowId);
    workflow.setStatus(Workflow.WorkflowStatus.RUNNING);

    when(executionDAOFacade.getTaskById(anyString())).thenReturn(maestroTask);
    when(executionDAOFacade.getWorkflowById(anyString(), anyBoolean())).thenReturn(workflow);

    maestroWorkflowExecutor.executeSystemTask(task2, taskId, 30);

    assertEquals(Task.Status.IN_PROGRESS, maestroTask.getStatus());
    assertEquals(1, maestroTask.getPollCount());
    verify(executionDAOFacade, times(1)).updateTask(any());
    assertTrue(maestroTask.getStartTime() > 0);
    assertEquals(30, maestroTask.getCallbackAfterSeconds());
  }