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);
  }
}

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 ) ; } }
wait for the next cluster state which satisfies changepredicate
