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);
            }
        }
    }
public void verifyIntegrity ( ) { LinkedHashSet < String > noLongerActive = new LinkedHashSet < String > ( myWorkUnits ) ; noLongerActive . removeAll ( allWorkUnits . keySet ( ) ) ; for ( String workUnit : noLongerActive ) { shutdownWork ( workUnit , true ) ; } 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.