code: public TaskStatus executeTask(ApplicationDefinition appDef, Task task) {
          checkServiceState();
          Tenant tenant = Tenant.getTenant(appDef);
          m_logger.debug("Checking that task {} in tenant {} is not running", task.getTaskID(), tenant);
          TaskRecord taskRecord = null;
          synchronized (m_executeLock) {
              taskRecord = waitForTaskStatus(tenant, task, s -> s != TaskStatus.IN_PROGRESS);
              taskRecord.setStatus(TaskStatus.NEVER_EXECUTED);
              updateTaskStatus(tenant, taskRecord, false);
              attemptToExecuteTask(appDef, task, taskRecord);
          }
          m_logger.debug("Checking that task {} in tenant {} has completed", tenant, task.getTaskID());
          taskRecord = waitForTaskStatus(tenant, task, s -> TaskStatus.isCompleted(s));
          return taskRecord.getStatus();
      }
docs: Execute the given task and return the task execution status. If the task is already
  running by this or another task manager, this method blocks until the task has
  finished. (TODO: should we put a timeout on this?) Once the task is not running, it
  then executed, even if we had to wait for a previous execution. In a multi-node
  cluster, the task may be executed by another node if just happened to be scheduled
  for execution, but in most cases, this task manager will execute the task.
  Regardless of whether this or a remote task manager executes the task, this method
  waits for that execution to complete and returns the final task status.
  
  @param appDef    {@link ApplicationDefinition} that defines the task's context
  including its tenant.
  @param task      Application-specific {@link Task} to execute.
  @return          Final {@link TaskStatus} of the task's execution.