public void verifyIntegrity ( ) { LinkedHashSet < String > noLongerActive = new LinkedHashSet < String > ( myWorkUnits ) ; noLongerActive . removeAll ( allWorkUnits . keySet ( ) ) ; for ( String workUnit : noLongerActive ) { shutdownWork ( workUnit , true ) ; } // Check the status of pegged work units to ensure that this node is not serving // a work unit that is pegged to another node in the cluster. for ( String workUnit : myWorkUnits ) { String claimPath = workUnitClaimPath ( workUnit ) ; if ( ! balancingPolicy . isFairGame ( workUnit ) && ! balancingPolicy . isPeggedToMe ( workUnit ) ) { LOG . info ( "Discovered I'm serving a work unit that's now " + "pegged to someone else. Shutting down {}" , workUnit ) ; shutdownWork ( workUnit , true ) ; } else if ( workUnitMap . containsKey ( workUnit ) && ! workUnitMap . get ( workUnit ) . equals ( myNodeID ) && ! claimedForHandoff . contains ( workUnit ) && ! znodeIsMe ( claimPath ) ) { LOG . info ( "Discovered I'm serving a work unit that's now " + "claimed by {} according to ZooKeeper. Shutting down {}" , workUnitMap . get ( workUnit ) , workUnit ) ; shutdownWork ( workUnit , true ) ; } } }
Verifies that all nodes are hooked up properly . Shuts down any work units which have been removed from the cluster or have been assigned to another node .