@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 shouldAggregateMetricsByQueryIdInSharedRuntimes() {
    // Given:
    final Map<String, String> sharedRuntimeQueryTags = ImmutableMap.of(
      "logical_cluster_id", "lksqlc-12345",
      "query-id", "CTAS_TEST_5",
      "member", "_confluent_blahblah_query-1-blahblah",
      "topic", TOPIC_NAME
    );
    listener.metricChange(mockMetric(
      BYTES_CONSUMED_TOTAL,
      2D,
      ImmutableMap.of(
        "thread-id", "_confluent_blahblah_query-1-blahblah",
        "task-id", "CTAS_TEST_5__" + TASK_ID_1,
        "processor-node-id", PROCESSOR_NODE_ID,
        "topic", TOPIC_NAME))
    );

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

    // When:
    listener.metricChange(mockMetric(
      BYTES_CONSUMED_TOTAL,
      15D,
      ImmutableMap.of(
        "thread-id", "_confluent_blahblah_query-1-blahblah",
        "task-id", "CTAS_TEST_5__" + TASK_ID_2,
        "processor-node-id", PROCESSOR_NODE_ID,
        "topic", TOPIC_NAME
      ))
    );

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

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