public void waitForSchemaAgreement(String targetSchemaVersion,int nodeCount){
  long start=System.currentTimeMillis();
  Map<String,List<String>> versions=null;
  while (System.currentTimeMillis() - start < MAX_SCHEMA_WAIT_MS) {
    log.info("schema version to sync to: {}, required node count: {}",targetSchemaVersion,nodeCount);
    versions=getSchemaVersions();
    if (versions.size() == 1) {
      if (!versions.containsKey(targetSchemaVersion)) {
        log.warn("Unable to converge to target version. Schema versions:{}, target version:{}",versions,targetSchemaVersion);
        return;
      }
      if (nodeCount != -1) {
        List<String> hosts=null;
        for (        Entry<String,List<String>> entry : versions.entrySet()) {
          hosts=entry.getValue();
        }
        if (hosts != null && hosts.size() == nodeCount) {
          log.info("schema versions converged to target version {}, required node count achieved: {}",targetSchemaVersion,nodeCount);
          return;
        }
      }
 else {
        log.info("schema versions converged to target version {}, no check for node count",targetSchemaVersion);
        return;
      }
    }
    log.info("waiting for schema change ...");
    try {
      Thread.sleep(SCHEMA_RETRY_SLEEP_MILLIS);
    }
 catch (    InterruptedException ex) {
    }
  }
  log.warn("Unable to converge schema versions: {}",versions);
}

public void waitForSchemaAgreement ( String targetSchemaVersion , int nodeCount ) { long start = System . currentTimeMillis ( ) ; Map < String , List < String > > versions = null ; while ( System . currentTimeMillis ( ) - start < MAX_SCHEMA_WAIT_MS ) { log . info ( "schema version to sync to: {}, required node count: {}" , targetSchemaVersion , nodeCount ) ; versions = getSchemaVersions ( ) ; if ( versions . size ( ) == 1 ) { if ( ! versions . containsKey ( targetSchemaVersion ) ) { log . warn ( "Unable to converge to target version. Schema versions:{}, target version:{}" , versions , targetSchemaVersion ) ; return ; } if ( nodeCount != - 1 ) { List < String > hosts = null ; for ( Entry < String , List < String > > entry : versions . entrySet ( ) ) { hosts = entry . getValue ( ) ; } if ( hosts != null && hosts . size ( ) == nodeCount ) { log . info ( "schema versions converged to target version {}, required node count achieved: {}" , targetSchemaVersion , nodeCount ) ; return ; } } else { log . info ( "schema versions converged to target version {}, no check for node count" , targetSchemaVersion ) ; return ; } } log . info ( "waiting for schema change ..." ) ; try { Thread . sleep ( SCHEMA_RETRY_SLEEP_MILLIS ) ; } catch ( InterruptedException ex ) { } } log . warn ( "Unable to converge schema versions: {}" , versions ) ; }
waits for schema change to propagate through all nodes of cluster it doesn ' t check if all nodes are of this version when nodecount = = - 1
