import com.stumbleupon.async.Deferred; //导入依赖的package包/类
@Test
public void runEmptyMetric() throws Exception {
  when(tsdb.addPoint(anyString(), anyLong(), anyLong(), anyMap()))
    .thenReturn(Deferred.fromResult(null));
  data = new Metric(null, TS, "42", TAGS);
  when(message.message()).thenReturn(JSON.serializeToBytes(data));
  KafkaRpcPluginThread writer = Mockito.spy(
      new KafkaRpcPluginThread(group, 1, TOPICS));

  writer.run();
  verify(tsdb, never()).addPoint(anyString(), anyLong(), anyLong(), anyMap());
  verify(tsdb, never()).addHistogramPoint(anyString(), anyLong(), 
      any(byte[].class), anyMap());
  verify(tsdb, never()).addAggregatePoint(anyString(), anyLong(), anyLong(), 
      anyMap(), anyBoolean(), anyString(), anyString(), anyString());
  verifyMessageRead(writer, false);
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
@Test
public void runStorageFailureRollup() throws Exception {
  when(group.getConsumerType()).thenReturn(TsdbConsumerType.ROLLUP);
  setupAggData(false, false);
  when(tsdb.addAggregatePoint(anyString(), anyLong(), anyLong(), anyMap(), 
      anyBoolean(), anyString(), anyString(), anyString()))
    .thenReturn(Deferred.fromResult(mock(HBaseException.class)));
  KafkaRpcPluginThread writer = Mockito.spy(
      new KafkaRpcPluginThread(group, 1, TOPICS));
  writer.run();

  verifyMessageRead(writer, true);
  verify(tsdb, times(1)).addAggregatePoint(METRIC, TS, 42L, TAGS, false, 
      "1h", "sum", null);
  verifyCtrsInc(new String[]{ "readRollupCounter", "requeuedRollupCounter",
    "storageExceptionCounter" });
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
@Test
public void runStorageFailurePreAgg() throws Exception {
  when(group.getConsumerType()).thenReturn(TsdbConsumerType.ROLLUP);
  setupAggData(false, true);
  when(tsdb.addAggregatePoint(anyString(), anyLong(), anyLong(), anyMap(), 
      anyBoolean(), anyString(), anyString(), anyString()))
    .thenReturn(Deferred.fromResult(mock(HBaseException.class)));
  KafkaRpcPluginThread writer = Mockito.spy(
      new KafkaRpcPluginThread(group, 1, TOPICS));
  writer.run();

  verifyMessageRead(writer, true);
  verify(tsdb, times(1)).addAggregatePoint(METRIC, TS, 42L, TAGS, true, 
      null, null, "sum");
  verifyCtrsInc(new String[]{ "readAggregateCounter", "requeuedAggregateCounter",
    "storageExceptionCounter" });
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
@Test
public void runStorageFailureHistogram() throws Exception {
  when(group.getConsumerType()).thenReturn(TsdbConsumerType.ROLLUP);
  setupHistogramData(false);
  when(tsdb.addHistogramPoint(anyString(), anyLong(), any(byte[].class), 
      anyMap()))
    .thenReturn(Deferred.fromResult(mock(HBaseException.class)));
  KafkaRpcPluginThread writer = Mockito.spy(
      new KafkaRpcPluginThread(group, 1, TOPICS));
  writer.run();

  verifyMessageRead(writer, true);
  verify(tsdb, times(1)).addHistogramPoint(eq(METRIC), eq(TS), 
      any(byte[].class), eq(TAGS));
  verifyCtrsInc(new String[]{ "readHistogramCounter", "requeuedHistogramCounter",
    "storageExceptionCounter" });
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
private void setupRawDataList(final boolean requeued) {
  when(tsdb.addPoint(anyString(), anyLong(), anyLong(), anyMap()))
          .thenReturn(Deferred.fromResult(null));

  TypedIncomingData ev1 = new Metric(METRIC, TS, "42", TAGS);
  TypedIncomingData ev2 = new Metric(METRIC2, TS, "42", TAGS);

  if (requeued) {
    ev1.setRequeueTS(TS + 60);
    ev2.setRequeueTS(TS + 60);
  }

  StringBuilder sb = new StringBuilder();
  sb.append("[");
  sb.append(new String(JSON.serializeToBytes(ev1)));
  sb.append(",");
  sb.append(new String(JSON.serializeToBytes(ev2)));
  sb.append("]");

  when(message.message()).thenReturn(sb.toString().getBytes());
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
private void setupHistogramData(final boolean requeued) {
  config.overrideConfig("tsd.core.histograms.config", 
      "{\"net.opentsdb.core.SimpleHistogramDecoder\": 42}");
  final HistogramCodecManager manager;
  manager = new HistogramCodecManager(tsdb);
  when(tsdb.histogramManager()).thenReturn(manager);
  when(tsdb.addHistogramPoint(anyString(), anyLong(), any(byte[].class), 
      anyMap()))
    .thenReturn(Deferred.fromResult(null));
  final Histogram histo = new Histogram();
  histo.setMetric(METRIC);
  histo.setTimestamp(TS);
  histo.setTags(new HashMap<String, String>(TAGS));
  histo.setOverflow(1);
  histo.setBuckets(ImmutableMap.<String, Long>builder()
      .put("0,1", 42L)
      .put("1,5", 24L)
      .build());
  data = histo;
  if (requeued) {
    data.setRequeueTS(TS + 60);
  }
  when(message.message()).thenReturn(JSON.serializeToBytes(data));
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
/**
 * {@inheritDoc}
 * @see net.opentsdb.tsd.RTPublisher#shutdown()
 */
@Override
public Deferred<Object> shutdown() {
	log.info(">>>>> Stopping TSDBChronicleEventPublisher");
	stopDisruptor("CacheLookup", cacheRbDisruptor);
	stopDisruptor("Dispatch", dispatchRbDisruptor);
	keepRunning.set(false);
	if(rolledFileDeletionThread!=null) rolledFileDeletionThread.interrupt();
	try { 
		outQueue.close();
		log.info("OutboundQueue Closed");
	} catch (Exception ex) {
		log.warn("Error closing OutboundQueue: {}", ex);
	}
	try {
		JMXHelper.unregisterMBean(JMXHelper.objectName("net.opentsdb:service=TSDBChronicleEventPublisher"));
	} catch (Exception x) {/* No Op */}

	log.info("<<<<< TSDBChronicleEventPublisher Stopped");
	return Deferred.fromResult(null);
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
Tagsets(AsyncKuduClient client, Tags tags, KuduTable tagsetsTable) {
  this.client = client;
  this.tagsetsTable = tagsetsTable;
  this.tags = tags;
  this.columnIndexes = ImmutableList.of(Tables.TAGSETS_ID_INDEX,
                                        Tables.TAGSETS_TAGSET_INDEX);
  this.tagsets = CacheBuilder.newBuilder()
      .maximumSize(1024 * 1024)
      .build(new CacheLoader<SerializedTagset, Deferred<Integer>>() {
        @Override
        public Deferred<Integer> load(SerializedTagset tagset) {
          return lookupOrInsertTagset(tagset, hashForTesting == null ?
                                              tagset.hashCode() : hashForTesting);
        }
      });
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
/**
 * Opens a Kudu TS instance on a Kudu cluster.
 *
 * @param kuduMasterAddressess list of "host:port" pair master addresses
 * @param name the name of the Kudu timeseries store. Multiple instances of
 *             Kudu TS can occupy the same Kudu cluster by using a different name.
 * @return the opened {@code KuduTS}.
 * @throws Exception on error
 */
public static KuduTS open(List<String> kuduMasterAddressess, String name) throws Exception {
  AsyncKuduClient client = new AsyncKuduClient.AsyncKuduClientBuilder(kuduMasterAddressess).build();

  Deferred<KuduTable> metricsDeferred = client.openTable(Tables.metricsTableName(name));
  Deferred<KuduTable> tagsetsDeferred = client.openTable(Tables.tagsetsTableName(name));
  Deferred<KuduTable> tagsDeferred = client.openTable(Tables.tagsTableName(name));
  KuduTable metricsTable = metricsDeferred.join(client.getDefaultAdminOperationTimeoutMs());
  KuduTable tagsetsTable = tagsetsDeferred.join(client.getDefaultAdminOperationTimeoutMs());
  KuduTable tagsTable = tagsDeferred.join(client.getDefaultAdminOperationTimeoutMs());

  Tags tags = new Tags(client, tagsTable);
  Tagsets tagsets = new Tagsets(client, tags, tagsetsTable);
  Metrics metrics = new Metrics(client, metricsTable, tagsets);
  return new KuduTS(client, name, metrics, tagsets, tags);
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
private static Deferred<KuduTable> openOrCreateTable(final AsyncKuduClient client,
                                                     final String table,
                                                     final Schema schema,
                                                     final CreateTableOptions options) throws Exception {
  class CreateTableErrback implements Callback<Deferred<KuduTable>, Exception> {
    @Override
    public Deferred<KuduTable> call(Exception e) throws Exception {
      // TODO(danburkert): we should only do this if the error is "not found"
      LOG.debug("Creating table {}", table);
      return client.createTable(table, schema, options);
    }
    @Override
    public String toString() {
      return MoreObjects.toStringHelper(this).add("table", table).toString();
    }
  }

  return client.openTable(table).addErrback(new CreateTableErrback());
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
@Test(timeout = 10000)
public void testConcurrentLookup() throws Exception {
  try (KuduTS ts = KuduTS.openOrCreate(ImmutableList.of(masterAddresses),
                                       "testConcurrentLookup",
                                       CreateOptions.defaults())) {
    Tagsets tagsets = ts.getTagsets();
    SortedMap<String, String> tagset = ImmutableSortedMap.of("k1", "v1");

    List<Deferred<Integer>> deferreds = new ArrayList<>();

    for (int i = 0; i < 10; i++) {
      deferreds.add(tagsets.getTagsetID(tagset));
      tagsets.clear();
    }

    assertEquals(1, ImmutableSet.copyOf(Deferred.group(deferreds).join()).size());
  }
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
@Test(timeout = 100000)
public void testHashCollisions() throws Exception {
  try (KuduTS ts = KuduTS.openOrCreate(ImmutableList.of(masterAddresses),
                                       "testHashCollisions",
                                       CreateOptions.defaults())) {
    Tagsets tagsets = ts.getTagsets();

    int numTagsets = 100;
    tagsets.setHashForTesting(0);

    List<Deferred<Integer>> deferreds = new ArrayList<>();
    for (int i = 0; i <= numTagsets; i++) {
      deferreds.add(tagsets.getTagsetID(ImmutableSortedMap.of("key", Integer.toString(i))));
    }

    List<Integer> ids = Deferred.group(deferreds).join();
    Collections.sort(ids);

    for (int i = 0; i < numTagsets; i++) {
      assertEquals(i, ids.get(i).intValue());
    }
  }
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
@Override
public void persist(String metric, DataPoint dp, Map<String, String> tags, final Function<Object, Void> callback) {
    try {
        if(callback == null) {
            tsdb.addPoint(metric, dp.getTimestamp(), dp.getValue(), tags).joinUninterruptibly();
        }
        else {

            Deferred<Object> ret = tsdb.addPoint(metric, dp.getTimestamp(), dp.getValue(), tags);
            ret.addCallback(new Callback<Object, Object>() {
                @Override
                public Object call(Object o) throws Exception {
                    return callback.apply(o);
                }
            });
        }
    } catch (Exception e) {
        throw new RuntimeException(e.getMessage(), e);
    }
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
/**
 * Flushes to BigTable any buffered client-side write operation.
 * <p>
 * @return A {@link Deferred}, whose callback chain will be invoked when
 * everything that was buffered at the time of the call has been flushed.
 * <p>
 * Note that this doesn't guarantee that <b>ALL</b> outstanding RPCs have
 * completed.  This doesn't introduce any sort of global sync point.  All
 * it does really is it sends any buffered RPCs to BigTable.
 */
public Deferred<Object> flush() {
  LOG.info("Flushing buffered mutations");
  final ArrayList<Deferred<Object>> deferreds = 
    new ArrayList<Deferred<Object>>(mutators.size());
  for (final BufferedMutator mutator : mutators.values()) {
    try {
      // TODO - run in a separate thread, breaks asynchronus behavior 
      // right now
      mutator.flush();
      deferreds.add(Deferred.fromResult(null));
    } catch (IOException e) {
      LOG.error("Error occurred while flushing buffer", e);
      deferreds.add(Deferred.fromError(e));
    }
  }
  num_flushes.increment();
  @SuppressWarnings("unchecked")
  final Deferred<Object> flushed = (Deferred) Deferred.group(deferreds);
  return flushed;
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
/**
 * Package-private access point for {@link Scanner}s to open themselves.
 * @param scanner The scanner to open.
 * @return A deferred scanner ID (long) if BigTable 0.94 and before, or a
 * deferred {@link Scanner.Response} if BigTable 0.95 and up.
 */
Deferred<Object> openScanner(final Scanner scanner) {
  num_scanners_opened.increment();

  if (LOG.isDebugEnabled()) {
    LOG.debug("BigTable API: Scanning table with {}", scanner.toString());
  }
  Table table = null;
  try {
    table = hbase_connection.getTable(TableName.valueOf(scanner.table()));
    ResultScanner result = table.getScanner(scanner.getHbaseScan());
    scanner.setResultScanner(result);
    scanner.setHbaseTable(table);

    return Deferred.fromResult(new Object());
  } catch (IOException e) {
    if (table != null) {
      try {
        table.close();
      } catch (Exception e1) {}
    }

    return Deferred.fromError(e);
  }
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
/**
 * Atomically and durably increments a value in BigTable.
 * <p>
 * This is equivalent to
 * {@link #atomicIncrement(AtomicIncrementRequest, boolean) atomicIncrement}
 * {@code (request, true)}
 * @param request The increment request.
 * @return The deferred {@code long} value that results from the increment.
 */
public Deferred<Long> atomicIncrement(final AtomicIncrementRequest request) {
  num_atomic_increments.increment();

  Table table = null;
  try {
    table = hbase_connection.getTable(TableName.valueOf(request.table()));
    long val = table.incrementColumnValue(request.key(),
            request.family(), request.qualifier(),
            request.getAmount(),
            request.isDurable() ? Durability.USE_DEFAULT : Durability.SKIP_WAL);

    LOG.info("BigTable API: AtomicIncrement for {} returned {}", request, val);
    return Deferred.fromResult(val);
  } catch (IOException e) {
    return Deferred.fromError(e);
  } finally {
    close(table);
  }
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
/** Scan with multiple qualifiers. */
@Test
public void scanWithQualifiers() throws Exception {
  client.setFlushInterval(FAST_FLUSH);
  final PutRequest put1 = new PutRequest(table, "k", family, "a", "val1");
  final PutRequest put2 = new PutRequest(table, "k", family, "b", "val2");
  final PutRequest put3 = new PutRequest(table, "k", family, "c", "val3");
  Deferred.group(client.put(put1), client.put(put2),
                 client.put(put3)).join();
  final Scanner scanner = client.newScanner(table);
  scanner.setFamily(family);
  scanner.setQualifiers(new byte[][] { { 'a' }, { 'c' } });
  final ArrayList<ArrayList<KeyValue>> rows = scanner.nextRows(2).join();
  assertSizeIs(1, rows);
  final ArrayList<KeyValue> kvs = rows.get(0);
  assertSizeIs(2, kvs);
  assertEq("val1", kvs.get(0).value());
  assertEq("val3", kvs.get(1).value());
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
/** Test regexp-based row key filtering.  */
@Test
public void keyRegexpFilter() throws Exception {
  client.setFlushInterval(FAST_FLUSH);
  final PutRequest put1 = new PutRequest(table, "krf accept:by the filter",
                                         family, "q", "krfv1");
  final PutRequest put2 = new PutRequest(table, "krf filtered out",
                                         family, "q", "krfv2");
  final PutRequest put3 = new PutRequest(table, "krf this is Accepted too",
                                         family, "q", "krfv3");
  Deferred.group(client.put(put1), client.put(put2),
                 client.put(put3)).join();
  final Scanner scanner = client.newScanner(table);
  scanner.setFamily(family);
  scanner.setStartKey("krf ");
  scanner.setStopKey("krf!");
  scanner.setKeyRegexp("[Aa]ccept(ed)?");
  final ArrayList<ArrayList<KeyValue>> rows = scanner.nextRows().join();
  assertSizeIs(2, rows);
  ArrayList<KeyValue> kvs = rows.get(0);
  assertSizeIs(1, kvs);
  assertEq("krfv1", kvs.get(0).value());
  kvs = rows.get(1);
  assertSizeIs(1, kvs);
  assertEq("krfv3", kvs.get(0).value());
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
/** Simple column prefix filter tests.  */
@Test
public void columnPrefixFilter() throws Exception {
  client.setFlushInterval(FAST_FLUSH);
  // Keep only rows with a column qualifier that starts with "qa".
  final PutRequest put1 = new PutRequest(table, "cpf1", family, "qa1", "v1");
  final PutRequest put2 = new PutRequest(table, "cpf1", family, "qa2", "v2");
  final PutRequest put3 = new PutRequest(table, "cpf2", family, "qa3", "v3");
  final PutRequest put4 = new PutRequest(table, "cpf2", family, "qb4", "v4");
  Deferred.group(Deferred.group(client.put(put1), client.put(put2)),
                 Deferred.group(client.put(put3), client.put(put4))).join();
  final Scanner scanner = client.newScanner(table);
  scanner.setFamily(family);
  scanner.setStartKey("cpf1");
  scanner.setStopKey("cpf3");
  scanner.setFilter(new ColumnPrefixFilter("qa"));
  final ArrayList<ArrayList<KeyValue>> rows = scanner.nextRows().join();
  assertSizeIs(2, rows);
  ArrayList<KeyValue> kvs = rows.get(0);
  assertSizeIs(2, kvs);
  assertEq("v1", kvs.get(0).value());
  assertEq("v2", kvs.get(1).value());
  kvs = rows.get(1);
  assertSizeIs(1, kvs);
  assertEq("v3", kvs.get(0).value());
}
 

import com.stumbleupon.async.Deferred; //导入依赖的package包/类
/** Simple column range filter tests.  */
@Test
public void columnRangeFilter() throws Exception {
  client.setFlushInterval(FAST_FLUSH);
  // Keep rows that have a qualifier in between "qb" (inclusive) and "qd4"
  // (exclusive).  So only v2 and v3 should be returned by the scanner.
  final PutRequest put1 = new PutRequest(table, "crf1", family, "qa1", "v1");
  final PutRequest put2 = new PutRequest(table, "crf1", family, "qb2", "v2");
  final PutRequest put3 = new PutRequest(table, "crf2", family, "qc3", "v3");
  final PutRequest put4 = new PutRequest(table, "crf2", family, "qd4", "v4");
  Deferred.group(Deferred.group(client.put(put1), client.put(put2)),
                 Deferred.group(client.put(put3), client.put(put4))).join();
  final Scanner scanner = client.newScanner(table);
  scanner.setFamily(family);
  scanner.setStartKey("crf1");
  scanner.setStopKey("crf3");
  scanner.setFilter(new ColumnRangeFilter("qb", true, "qd4", false));
  final ArrayList<ArrayList<KeyValue>> rows = scanner.nextRows().join();
  assertSizeIs(2, rows);  // One KV from row "fl1" and one from "fl2".
  ArrayList<KeyValue> kvs = rows.get(0);
  assertSizeIs(1, kvs);
  assertEq("v2", kvs.get(0).value());
  kvs = rows.get(1);
  assertSizeIs(1, kvs);
  assertEq("v3", kvs.get(0).value());
}