public Table<Locator, String, String> getMetadataValues(Set<Locator> locators) {
        ColumnFamily CF = CassandraModel.CF_METRIC_METADATA;
        boolean isBatch = locators.size() > 1;
        Table<Locator, String, String> metaTable = HashBasedTable.create();
        Timer.Context ctx = isBatch ? Instrumentation.getBatchReadTimerContext(CF) : Instrumentation.getReadTimerContext(CF);
        try {
            // We don't paginate this call. So we should make sure the number of reads is tolerable.
            // TODO: Think about paginating this call.
            OperationResult<Rows<Locator, String>> query = keyspace
                    .prepareQuery(CF)
                    .getKeySlice(locators)
                    .execute();
            for (Row<Locator, String> row : query.getResult()) {
                ColumnList<String> columns = row.getColumns();
                for (Column<String> column : columns) {
                    String metaValue = column.getValue(StringMetadataSerializer.get());
                    String metaKey = column.getName();
                    metaTable.put(row.getKey(), metaKey, metaValue);
                }
            }
        } catch (ConnectionException e) {
            if (e instanceof NotFoundException) { // TODO: Not really sure what happens when one of the keys is not found.
                Instrumentation.markNotFound(CF);
            } else {
                if (isBatch) { Instrumentation.markBatchReadError(e); }
                else { Instrumentation.markReadError(e); }
            }
            log.warn((isBatch ? "Batch " : "") + " read query failed for column family " + CF.getName(), e);
        } finally {
            ctx.stop();
        }
        return metaTable;
    }
public Table<Locator, String, String> getMetadataValues(Set<Locator> locators) {
        ColumnFamily CF = CassandraModel.CF_METRIC_METADATA;
        boolean isBatch = locators.size() > 1;
        Table<Locator, String, String> metaTable = HashBasedTable.create();
        Timer.Context ctx = isBatch ? Instrumentation.getBatchReadTimerContext(CF) : Instrumentation.getReadTimerContext(CF);
        try {
           
           
            OperationResult<Rows<Locator, String>> query = keyspace
                    .prepareQuery(CF)
                    .getKeySlice(locators)
                    .execute();
            for (Row<Locator, String> row : query.getResult()) {
                ColumnList<String> columns = row.getColumns();
                for (Column<String> column : columns) {
                    String metaValue = column.getValue(StringMetadataSerializer.get());
                    String metaKey = column.getName();
                    metaTable.put(row.getKey(), metaKey, metaValue);
                }
            }
        } catch (ConnectionException e) {
            if (e instanceof NotFoundException) {
                Instrumentation.markNotFound(CF);
            } else {
                if (isBatch) { Instrumentation.markBatchReadError(e); }
                else { Instrumentation.markReadError(e); }
            }
            log.warn((isBatch ? "Batch " : "") + " read query failed for column family " + CF.getName(), e);
        } finally {
            ctx.stop();
        }
        return metaTable;
    }
public table<locator, string, string> getmetadatavalues(set<locator> locators) { columnfamily cf = cassandramodel.cf_metric_metadata; boolean isbatch = locators.size() > 1; table<locator, string, string> metatable = hashbasedtable.create(); timer.context ctx = isbatch ? instrumentation.getbatchreadtimercontext(cf) : instrumentation.getreadtimercontext(cf); try { operationresult<rows<locator, string>> query = keyspace .preparequery(cf) .getkeyslice(locators) .execute(); for (row<locator, string> row : query.getresult()) { columnlist<string> columns = row.getcolumns(); for (column<string> column : columns) { string metavalue = column.getvalue(stringmetadataserializer.get()); string metakey = column.getname(); metatable.put(row.getkey(), metakey, metavalue); } } } catch (connectionexception e) { if (e instanceof notfoundexception) { instrumentation.marknotfound(cf); } else { if (isbatch) { instrumentation.markbatchreaderror(e); } else { instrumentation.markreaderror(e); } } log.warn((isbatch ? "batch " : "") + " read query failed for column family " + cf.getname(), e); } finally { ctx.stop(); } return metatable; }