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();
        }
      }
    }
  }
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 { 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 ( ) ) ; runPendingTasks ( ) ; } } } }
Ensures no workers are already running a task before assigning the task to a worker.
It is possible that a worker is running a task that the RTR has no knowledge of. This occurs when the RTR
needs to bootstrap after a restart.

@param taskRunnerWorkItem - the task to assign

@return true iff the task is now assigned