@ Override public void claimWork ( ) throws InterruptedException { synchronized ( cluster . allWorkUnits ) { for ( String workUnit : getUnclaimed ( ) ) { if ( isPeggedToMe ( workUnit ) ) { claimWorkPeggedToMe ( workUnit ) ; } } final double evenD = evenDistribution ( ) ; LinkedList < String > unclaimed = new LinkedList < String > ( getUnclaimed ( ) ) ; while ( myLoad ( ) <= evenD && ! unclaimed . isEmpty ( ) ) { final String workUnit = unclaimed . poll ( ) ; if ( config . useSoftHandoff && cluster . containsHandoffRequest ( workUnit ) && isFairGame ( workUnit ) && attemptToClaim ( workUnit , true ) ) { LOG . info ( workUnit ) ; handoffListener . finishHandoff ( workUnit ) ; } else if ( isFairGame ( workUnit ) ) { attemptToClaim ( workUnit ) ; } } } }
Begins by claiming all work units that are pegged to this node . Then continues to claim work from the available pool until we ve claimed equal to or slightly more than the total desired load .