public void stopAndWait ( final long waitTime , final AtomicBoolean stopFlag ) { if ( ! waitInProgress . getAndSet ( true ) ) { stopFlag . set ( true ) ; rejoinExecutor . submit ( new Runnable ( ) { @ Override public void run ( ) { balancingPolicy . drainToCount ( 0 , false ) ; try { Thread . sleep ( waitTime ) ; } catch ( InterruptedException e ) { LOG . warn ( "Interrupted while waiting." ) ; } LOG . info ( "Back to work." ) ; stopFlag . set ( false ) ; waitInProgress . set ( false ) ; } } ) ; } }
For handling problematic nodes - drains workers and does not claim work for waitTime seconds