@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 shouldAddNewMeasurableForAllThroughputMetricsTypes() {
    // Given:
    listener.metricChange(mockMetric(
        BYTES_CONSUMED_TOTAL,
        1D,
        STREAMS_TAGS_TASK_1)
    );

    listener.metricChange(mockMetric(
        RECORDS_CONSUMED_TOTAL,
        2D,
        STREAMS_TAGS_TASK_1)
    );

    listener.metricChange(mockMetric(
        BYTES_PRODUCED_TOTAL,
        3D,
        STREAMS_TAGS_TASK_1)
    );

    listener.metricChange(mockMetric(
        RECORDS_PRODUCED_TOTAL,
        4D,
        STREAMS_TAGS_TASK_1)
    );

    // When:
    final Measurable bytesConsumed = verifyAndGetMetric(BYTES_CONSUMED_TOTAL, QUERY_ONE_TAGS);
    final Measurable recordsConsumed = verifyAndGetMetric(RECORDS_CONSUMED_TOTAL, QUERY_ONE_TAGS);
    final Measurable bytesProduced = verifyAndGetMetric(BYTES_PRODUCED_TOTAL, QUERY_ONE_TAGS);
    final Measurable recordsProduced = verifyAndGetMetric(RECORDS_PRODUCED_TOTAL, QUERY_ONE_TAGS);

    // Then:
    assertThat(bytesConsumed.measure(new MetricConfig().tags(QUERY_ONE_TAGS), 0L), equalTo(1D));
    assertThat(recordsConsumed.measure(new MetricConfig().tags(QUERY_ONE_TAGS), 0L), equalTo(2D));
    assertThat(bytesProduced.measure(new MetricConfig().tags(QUERY_ONE_TAGS), 0L), equalTo(3D));
    assertThat(recordsProduced.measure(new MetricConfig().tags(QUERY_ONE_TAGS), 0L), equalTo(4D));
  }