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 ; } }
Creates a ZK entry under a specific path associated with a worker . The worker is responsible for removing the task ZK entry and creating a task status ZK entry .