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 ( ) ; } } } }
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 .