docs: private boolean tryAssignTask(final Task task, final RemoteTaskRunnerWorkItem taskRunnerWorkItem) throws Exception
    {
      Preconditions.checkNotNull(task, "task");
      Preconditions.checkNotNull(taskRunnerWorkItem, "taskRunnerWorkItem");
      Preconditions.checkArgument(task.getId().equals(taskRunnerWorkItem.getTaskId()), "task id != workItem id");
  
      if (runningTasks.containsKey(task.getId()) || findWorkerRunningTask(task.getId()) != null) {
        log.info("Task[%s] already running.", task.getId());
        return true;
      } else {
        // Nothing running this task, announce it in ZK for a worker to run it
        WorkerBehaviorConfig workerConfig = workerConfigRef.get();
        WorkerSelectStrategy strategy;
        if (workerConfig == null || workerConfig.getSelectStrategy() == null) {
          strategy = WorkerBehaviorConfig.DEFAULT_STRATEGY;
          log.debug("No worker selection strategy set. Using default of [%s]", strategy.getClass().getSimpleName());
        } else {
          strategy = workerConfig.getSelectStrategy();
        }
  
        ZkWorker assignedWorker = null;
        final ImmutableWorkerInfo immutableZkWorker;
        try {
          synchronized (workersWithUnacknowledgedTask) {
            immutableZkWorker = strategy.findWorkerForTask(
                config,
                ImmutableMap.copyOf(
                    Maps.transformEntries(
                        Maps.filterEntries(
                            zkWorkers,
                            new Predicate<Map.Entry<String, ZkWorker>>()
                            {
                              @Override
                              public boolean apply(Map.Entry<String, ZkWorker> input)
                              {
                                return !lazyWorkers.containsKey(input.getKey()) &&
                                       !workersWithUnacknowledgedTask.containsKey(input.getKey()) &&
                                       !blackListedWorkers.contains(input.getValue());
                              }
                            }
                        ),
                        (String key, ZkWorker value) -> value.toImmutable()
                    )
                ),
                task
            );
  
            if (immutableZkWorker != null &&
                workersWithUnacknowledgedTask.putIfAbsent(immutableZkWorker.getWorker().getHost(), task.getId())
                == null) {
              assignedWorker = zkWorkers.get(immutableZkWorker.getWorker().getHost());
            }
          }
  
          if (assignedWorker != null) {
            return announceTask(task, assignedWorker, taskRunnerWorkItem);
          } else {
            log.debug(
                "Unsuccessful task-assign attempt for task [%s] on workers [%s]. Workers to ack tasks are [%s].",
                task.getId(),
                zkWorkers.values(),
                workersWithUnacknowledgedTask
            );
          }
  
          return false;
        }
        finally {
          if (assignedWorker != null) {
            workersWithUnacknowledgedTask.remove(assignedWorker.getWorker().getHost());
            //if this attempt won the race to run the task then other task might be able to use this worker now after task ack.
            runPendingTasks();
          }
        }
      }
    }