@ 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 ( ) ; } }
A task will be run only if there is no current knowledge in the RemoteTaskRunner of the task .