public static void validateSchemaIsSettled(Cassandra.Client thriftClient,
String currentVersionId) throws InterruptedException, StorageException {
log.debug("Waiting for Cassandra schema propagation...");
Map<String, List<String>> versions = null;
final TimeUUIDType ti = TimeUUIDType.instance;
final long start = System.currentTimeMillis();
long lastTry = 0;
final long limit = start + SCHEMA_WAIT_MAX;
final long minSleep = SCHEMA_WAIT_INCREMENT;
boolean inAgreement = false;
outer:
while (limit - System.currentTimeMillis() >= 0 && !inAgreement) {
// Block for a little while if we're looping too fast
final long now = System.currentTimeMillis();
long sinceLast = now - lastTry;
long willSleep = minSleep - sinceLast;
if (0 < willSleep) {
log.debug("Schema not yet propagated; " +
"rechecking in {} ms", willSleep);
Thread.sleep(willSleep);
}
// Issue thrift query
try {
lastTry = System.currentTimeMillis();
versions = thriftClient.describe_schema_versions(); // getting schema version for nodes of the ring
} catch (Exception e) {
throw new PermanentStorageException("Failed to fetch Cassandra Thrift schema versions: " +
((e instanceof InvalidRequestException) ?
((InvalidRequestException) e).getWhy() : e.getMessage()));
}
int nodeCount = 0;
// Check schema version
UUID benchmark = UUID.fromString(currentVersionId);
ByteBuffer benchmarkBB = ti.decompose(benchmark);
for (String version : versions.keySet()) {
if (version.equals(StorageProxy.UNREACHABLE)) {
nodeCount += versions.get(version).size();
continue;
}
UUID uuid = UUID.fromString(version);
ByteBuffer uuidBB = ti.decompose(uuid);
if (-1 < ti.compare(uuidBB, benchmarkBB)) {
log.debug("Version {} equals or comes after required version {}", uuid, benchmark);
nodeCount += versions.get(version).size();
continue;
}
continue outer;