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

    Measurable bytesConsumed = verifyAndGetMetric(BYTES_CONSUMED_TOTAL, QUERY_ONE_TAGS);
    Object bytesConsumedValue = bytesConsumed.measure(new MetricConfig().tags(QUERY_ONE_TAGS), 0L);
    assertThat(bytesConsumedValue, equalTo(2D));

    // When:
    listener.metricChange(mockMetric(
        BYTES_CONSUMED_TOTAL,
        15D,
        STREAMS_TAGS_TASK_1)
    );

    // Then:
    bytesConsumed = verifyAndGetMetric(BYTES_CONSUMED_TOTAL, QUERY_ONE_TAGS);
    bytesConsumedValue = bytesConsumed.measure(new MetricConfig().tags(QUERY_ONE_TAGS), 0L);

    assertThat(bytesConsumedValue, equalTo(15D));
  }