docs: private boolean announceTask(
        final Task task,
        final ZkWorker theZkWorker,
        final RemoteTaskRunnerWorkItem taskRunnerWorkItem
    ) throws Exception
    {
      Preconditions.checkArgument(task.getId().equals(taskRunnerWorkItem.getTaskId()), "task id != workItem id");
      final String worker = theZkWorker.getWorker().getHost();
      synchronized (statusLock) {
        if (!zkWorkers.containsKey(worker) || lazyWorkers.containsKey(worker)) {
          // the worker might have been killed or marked as lazy
          log.info("Not assigning task to already removed worker[%s]", worker);
          return false;
        }
        log.info("Coordinator asking Worker[%s] to add task[%s]", worker, task.getId());
  
        CuratorUtils.createIfNotExists(
            cf,
            JOINER.join(indexerZkConfig.getTasksPath(), worker, task.getId()),
            CreateMode.EPHEMERAL,
            jsonMapper.writeValueAsBytes(task),
            config.getMaxZnodeBytes()
        );
  
        RemoteTaskRunnerWorkItem workItem = pendingTasks.remove(task.getId());
        if (workItem == null) {
          log.makeAlert("WTF?! Got a null work item from pending tasks?! How can this be?!")
             .addData("taskId", task.getId())
             .emit();
          return false;
        }
  
        RemoteTaskRunnerWorkItem newWorkItem = workItem.withWorker(theZkWorker.getWorker(), null);
        runningTasks.put(task.getId(), newWorkItem);
        log.info("Task %s switched from pending to running (on [%s])", task.getId(), newWorkItem.getWorker().getHost());
        TaskRunnerUtils.notifyStatusChanged(listeners, task.getId(), TaskStatus.running(task.getId()));
  
        // Syncing state with Zookeeper - don't assign new tasks until the task we just assigned is actually running
        // on a worker - this avoids overflowing a worker with tasks
        Stopwatch timeoutStopwatch = Stopwatch.createStarted();
        while (!isWorkerRunningTask(theZkWorker, task.getId())) {
          final long waitMs = config.getTaskAssignmentTimeout().toStandardDuration().getMillis();
          statusLock.wait(waitMs);
          long elapsed = timeoutStopwatch.elapsed(TimeUnit.MILLISECONDS);
          if (elapsed >= waitMs) {
            log.makeAlert(
                "Task assignment timed out on worker [%s], never ran task [%s]! Timeout: (%s >= %s)!",
                worker,
                task.getId(),
                elapsed,
                config.getTaskAssignmentTimeout()
            ).emit();
            taskComplete(taskRunnerWorkItem, theZkWorker, TaskStatus.failure(task.getId()));
            break;
          }
        }
        return true;
      }
    }