question: public void waitForNextChange ( Listener listener , ChangePredicate changePredicate , @ Nullable TimeValue timeOutValue ) { if ( observingContext . get ( ) != null ) { throw new ElasticsearchException ( "already waiting for a cluster state change" ) ; } Long timeoutTimeLeftMS ; if ( timeOutValue == null ) { timeOutValue = this . timeOutValue ; if ( timeOutValue != null ) { long timeSinceStartMS = TimeValue . nsecToMSec ( System . nanoTime ( ) - startTimeNS ) ; timeoutTimeLeftMS = timeOutValue . millis ( ) - timeSinceStartMS ; if ( timeoutTimeLeftMS <= 0l ) { logger . trace ( "observer timed out. notifying listener. timeout setting [{}], time since start [{}]" , timeOutValue , new TimeValue ( timeSinceStartMS ) ) ; timedOut = true ; lastObservedState . set ( new ObservedState ( clusterService . state ( ) ) ) ; listener . onTimeout ( timeOutValue ) ; return ; } } else { timeoutTimeLeftMS = null ; } } else { this . startTimeNS = System . nanoTime ( ) ; this . timeOutValue = timeOutValue ; timeoutTimeLeftMS = timeOutValue . millis ( ) ; timedOut = false ; } ObservedState newState = new ObservedState ( clusterService . state ( ) ) ; ObservedState lastState = lastObservedState . get ( ) ; if ( changePredicate . apply ( lastState . clusterState , lastState . status , newState . clusterState , newState . status ) ) { logger . trace ( "observer: sampled state accepted by predicate ({})" , newState ) ; lastObservedState . set ( newState ) ; listener . onNewClusterState ( newState . clusterState ) ; } else { logger . trace ( "observer: sampled state rejected by predicate ({}). adding listener to ClusterService" , newState ) ; ObservingContext context = new ObservingContext ( listener , changePredicate ) ; if ( ! observingContext . compareAndSet ( null , context ) ) { throw new ElasticsearchException ( "already waiting for a cluster state change" ) ; } clusterService . add ( timeoutTimeLeftMS == null ? null : new TimeValue ( timeoutTimeLeftMS ) , clusterStateListener ) ; } }
target:  Wait for the next cluster state which satisfies changePredicate