StormBenchmark implements IBenchmark { @Override public IMetricsCollector getMetricsCollector(Config config, StormTopology topology) { Set<MetricsItem> items = Sets.newHashSet( MetricsItem.SUPERVISOR_STATS, MetricsItem.TOPOLOGY_STATS, MetricsItem.THROUGHPUT, MetricsItem.SPOUT_THROUGHPUT, MetricsItem.SPOUT_LATENCY ); return new BasicMetricsCollector(config, topology, items); }  @Override IMetricsCollector getMetricsCollector(Config config, StormTopology topology); static final String DEFAULT_TOPOLOGY_NAME; }
@Test(dataProvider = "getStormBenchmark") public void getMetricsCollectorShouldReturnProperCollector( StormBenchmark benchmark, Class collectorClass) { IMetricsCollector collector = benchmark.getMetricsCollector(config, topology); assertThat(collector) .isNotNull() .isInstanceOf(collectorClass); }