public void finishHandoff ( final String workUnit ) throws InterruptedException { String unitId = cluster . workUnitMap . get ( workUnit ) ; LOG . info ( "Handoff of {} to me acknowledged. Deleting claim ZNode for {} and waiting for {} to shutdown work." , workUnit , workUnit , ( ( unitId == null ) ? "(None)" : unitId ) ) ; final String path = cluster . workUnitClaimPath ( workUnit ) ; Stat stat = ZKUtils . exists ( cluster . zk , path , new Watcher ( ) { @ Override public void process ( WatchedEvent event ) { // Don't really care about the type of event here - call unconditionally to clean up state completeHandoff ( workUnit , path ) ; } } ) ; // Unlikely that peer will have already deleted znode, but handle it regardless if ( stat == null ) { LOG . warn ( "Peer already deleted znode of {}" , workUnit ) ; completeHandoff ( workUnit , path ) ; } }
Completes the process of handing off a work unit from one node to the current one . Attempts to establish a final claim to the node handed off to me in ZooKeeper and repeats execution of the task every two seconds until it is complete .