@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());
    // also verification
    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);
}