protected void drainToCount ( final int targetCount , final boolean doShutdown , /* Boolean = false */ final boolean useHandoff , /* : Boolean = config.useSoftHandoff */ final CountDownLatch latch ) { String msg = useHandoff ? " with handoff" : "" ; LOG . info ( String . format ( "Draining %s%s. Target count: %s, Current: %s" , config . workUnitName , msg , targetCount , cluster . myWorkUnits . size ( ) ) ) ; if ( targetCount >= cluster . myWorkUnits . size ( ) ) { if ( ! doShutdown ) { return ; } } else if ( targetCount == 0 && doShutdown ) { cluster . completeShutdown ( ) ; } final int amountToDrain = cluster . myWorkUnits . size ( ) - targetCount ; String msgPrefix = ( useHandoff ) ? "Requesting handoff for" : "Shutting down" ; LOG . info ( "{} {} of {} {} over {} seconds" , msgPrefix , amountToDrain , cluster . myWorkUnits . size ( ) , config . workUnitName , config . drainTime ) ; // Build a list of work units to hand off. LinkedHashSet < String > wuList = new LinkedHashSet < String > ( cluster . myWorkUnits ) ; wuList . removeAll ( cluster . workUnitsPeggedToMe ) ; final ArrayList < String > toHandOff = new ArrayList < String > ( ) ; Iterator < String > it = wuList . iterator ( ) ; for ( int i = amountToDrain ; it . hasNext ( ) && i > 0 ; -- i ) { toHandOff . add ( it . next ( ) ) ; } final int drainInterval = ( int ) ( ( ( double ) config . drainTime / ( double ) toHandOff . size ( ) ) * 1000 ) ; final Iterator < String > drainIt = toHandOff . iterator ( ) ; TimerTask handoffTask = new TimerTask ( ) { @ Override public void run ( ) { if ( ! drainIt . hasNext ( ) ) { if ( targetCount == 0 && doShutdown ) { cluster . completeShutdown ( ) ; } if ( latch != null ) { latch . countDown ( ) ; } return ; } String workUnit = drainIt . next ( ) ; if ( useHandoff && ! isPeggedToMe ( workUnit ) ) { try { cluster . requestHandoff ( workUnit ) ; } catch ( Exception e ) { LOG . warn ( "Problems trying to request handoff of " + workUnit , e ) ; } } else { cluster . shutdownWork ( workUnit , true ) ; } cluster . schedule ( this , drainInterval , TimeUnit . MILLISECONDS ) ; } } ; LOG . info ( "Releasing {} / {} work units over {} seconds: {}" , amountToDrain , cluster . myWorkUnits . size ( ) , config . drainTime , Strings . mkstring ( toHandOff , ", " ) ) ; if ( ! cluster . myWorkUnits . isEmpty ( ) ) { cluster . schedule ( handoffTask , 0 , TimeUnit . SECONDS ) ; } }
Drains this node s share of the cluster workload down to a specific number of work units over a period of time specified in the configuration with soft handoff if enabled ..