@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 shouldAggregateOverAllTasksAndProcessorNodes() {
    // Given:
    KafkaMetric m1 = mockMetric(
        BYTES_CONSUMED_TOTAL,
        2D,
        STREAMS_TAGS_TASK_1
    );
    KafkaMetric m2 = mockMetric(
        BYTES_CONSUMED_TOTAL,
        5D,
        STREAMS_TAGS_TASK_2
    );
    KafkaMetric m3 = mockMetric(
        BYTES_CONSUMED_TOTAL,
        3D,
        STREAMS_TAGS_PROCESSOR_2
    );

    // When:
    listener.metricChange(m1);
    listener.metricChange(m2);
    listener.metricChange(m3);

    // Then:
    final Measurable bytesConsumed = verifyAndGetMetric(BYTES_CONSUMED_TOTAL, QUERY_ONE_TAGS);
    final Object value = bytesConsumed.measure(new MetricConfig().tags(QUERY_ONE_TAGS), 0L);
    assertThat(value, equalTo(10D));
  }