StormTopologyTimeSeriesMetricsImpl implements TopologyTimeSeriesMetrics { @Override public Map<String, Map<Long, Double>> getkafkaTopicOffsets(TopologyLayout topology, Component component, long from, long to, String asUser) { String stormTopologyName = StormTopologyUtil.findOrGenerateTopologyName(client, topology.getId(), topology.getName(), asUser); String stormComponentName = getComponentName(component); String topicName = findKafkaTopicName(component); if (topicName == null) { throw new IllegalStateException("Cannot find Kafka topic name from source config - topology name: " + topology.getName() + " / source : " + component.getName()); } StormMappedMetric[] metrics = { StormMappedMetric.logsize, StormMappedMetric.offset, StormMappedMetric.lag }; Map<String, Map<Long, Double>> kafkaOffsets = new HashMap<>(); for (StormMappedMetric metric : metrics) { kafkaOffsets.put(metric.name(), queryKafkaMetrics(stormTopologyName, stormComponentName, metric, topicName, from, to)); } return kafkaOffsets; } 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 getKafkaTopicOffsets() throws Exception { final long from = 1L; final long to = 3L; final Map<String, Map<Long, Double>> expected = new HashMap<>(); expected.put(StormMappedMetric.logsize.name(), generateTestPointsMap()); expected.put(StormMappedMetric.offset.name(), generateTestPointsMap()); expected.put(StormMappedMetric.lag.name(), generateTestPointsMap()); new Expectations() {{ mockTimeSeriesQuerier.getMetrics( withEqual(mockedTopologyName), withEqual(source.getId() + "-" + source.getName()), withEqual(String.format(StormMappedMetric.logsize.getStormMetricName(), TOPIC_NAME)), withEqual(StormMappedMetric.logsize.getAggregateFunction()), withEqual(from), withEqual(to) ); result = expected.get(StormMappedMetric.logsize.name()); mockTimeSeriesQuerier.getMetrics( withEqual(mockedTopologyName), withEqual(source.getId() + "-" + source.getName()), withEqual(String.format(StormMappedMetric.offset.getStormMetricName(), TOPIC_NAME)), withEqual(StormMappedMetric.offset.getAggregateFunction()), withEqual(from), withEqual(to) ); result = expected.get(StormMappedMetric.offset.name()); mockTimeSeriesQuerier.getMetrics( withEqual(mockedTopologyName), withEqual(source.getId() + "-" + source.getName()), withEqual(String.format(StormMappedMetric.lag.getStormMetricName(), TOPIC_NAME)), withEqual(StormMappedMetric.lag.getAggregateFunction()), withEqual(from), withEqual(to) ); result = expected.get(StormMappedMetric.lag.name()); }}; Map<String, Map<Long, Double>> actual = stormTopologyTimeSeriesMetrics.getkafkaTopicOffsets(topology, source, from, to, null); assertEquals(expected, actual); }