private void cleanup ( final String taskId ) { if ( ! lifecycleLock . awaitStarted ( 1 , TimeUnit . SECONDS ) ) { return ; } final RemoteTaskRunnerWorkItem removed = completeTasks . remove ( taskId ) ; final Worker worker = removed . getWorker ( ) ; if ( removed == null || worker == null ) { log . makeAlert ( "WTF?! Asked to cleanup nonexistent task" ) . addData ( "taskId" , taskId ) . emit ( ) ; } else { final String workerId = worker . getHost ( ) ; log . info ( "Cleaning up task[%s] on worker[%s]" , taskId , workerId ) ; final String statusPath = JOINER . join ( indexerZkConfig . getStatusPath ( ) , workerId , taskId ) ; try { cf . delete ( ) . guaranteed ( ) . forPath ( statusPath ) ; } catch ( KeeperException . NoNodeException e ) { log . info ( "Tried to delete status path[%s] that didn't exist! Must've gone away already?" , statusPath ) ; } catch ( Exception e ) { throw new RuntimeException ( e ) ; } } }
Removes a task from the complete queue and clears out the ZK status path of the task .