docs: @Override
    public ListenableFuture<TaskStatus> run(final Task task)
    {
      final RemoteTaskRunnerWorkItem completeTask, runningTask, pendingTask;
      if ((pendingTask = pendingTasks.get(task.getId())) != null) {
        log.info("Assigned a task[%s] that is already pending!", task.getId());
        runPendingTasks();
        return pendingTask.getResult();
      } else if ((runningTask = runningTasks.get(task.getId())) != null) {
        ZkWorker zkWorker = findWorkerRunningTask(task.getId());
        if (zkWorker == null) {
          log.warn("Told to run task[%s], but no worker has started running it yet.", task.getId());
        } else {
          log.info("Task[%s] already running on %s.", task.getId(), zkWorker.getWorker().getHost());
          TaskAnnouncement announcement = zkWorker.getRunningTasks().get(task.getId());
          if (announcement.getTaskStatus().isComplete()) {
            taskComplete(runningTask, zkWorker, announcement.getTaskStatus());
          }
        }
        return runningTask.getResult();
      } else if ((completeTask = completeTasks.get(task.getId())) != null) {
        return completeTask.getResult();
      } else {
        return addPendingTask(task).getResult();
      }
    }