StormTopologyTimeSeriesMetricsImpl implements TopologyTimeSeriesMetrics { @Override public Map<Long, Double> getCompleteLatency(TopologyLayout topology, Component component, long from, long to, String asUser) { String stormTopologyName = StormTopologyUtil.findOrGenerateTopologyName(client, topology.getId(), topology.getName(), asUser); String stormComponentName = getComponentName(component); return queryComponentMetrics(stormTopologyName, stormComponentName, StormMappedMetric.completeLatency, from, to); } StormTopologyTimeSeriesMetricsImpl(StormRestAPIClient client); @Override void setTimeSeriesQuerier(TimeSeriesQuerier timeSeriesQuerier); @Override TimeSeriesQuerier getTimeSeriesQuerier(); @Override Map<Long, Double> getCompleteLatency(TopologyLayout topology, Component component, long from, long to, String asUser); @Override Map<String, Map<Long, Double>> getkafkaTopicOffsets(TopologyLayout topology, Component component, long from, long to, String asUser); @Override TimeSeriesComponentMetric getTopologyStats(TopologyLayout topology, long from, long to, String asUser); @Override TimeSeriesComponentMetric getComponentStats(TopologyLayout topology, Component component, long from, long to, String asUser);  }
@Test public void testGetCompleteLatency() throws Exception { final long from = 1L; final long to = 3L; final Map<Long, Double> expected = generateTestPointsMap(); new Expectations() {{ mockTimeSeriesQuerier.getMetrics( withEqual(mockedTopologyName), withEqual(source.getId() + "-" + source.getName()), withEqual(StormMappedMetric.completeLatency.getStormMetricName()), withEqual(StormMappedMetric.completeLatency.getAggregateFunction()), withEqual(from), withEqual(to) ); result = expected; }}; Map<Long, Double> actual = stormTopologyTimeSeriesMetrics.getCompleteLatency(topology, source, from, to, null); assertEquals(expected, actual); }