@Override
  public void metricChange(final KafkaMetric metric) {
    if (!THROUGHPUT_METRIC_NAMES.contains(metric.metricName().name())
          || !StreamsMetricsImpl.TOPIC_LEVEL_GROUP.equals(metric.metricName().group())) {
      return;
    }
    addMetric(
        metric,
        getQueryId(metric),
        getTopic(metric)
    );
  }
@Test
  public void shouldIgnoreNonThroughputMetric() {
    // When:
    listener.metricChange(mockMetric(
      "other-metric",
      2D,
      STREAMS_TAGS_TASK_1)
    );

    // Then:
    assertThrows(AssertionError.class, () -> verifyAndGetMetric("other-metric", QUERY_ONE_TAGS));
  }