@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 shouldAggregateMetricsByQueryIdForTransientQueries() {
    // Given:
    final Map<String, String> transientQueryTags = ImmutableMap.of(
        "logical_cluster_id", "lksqlc-12345",
        "query-id", "blahblah_4",
        "member", TRANSIENT_THREAD_ID,
        "topic", TOPIC_NAME
    );
    listener.metricChange(mockMetric(
        BYTES_CONSUMED_TOTAL,
        2D,
        ImmutableMap.of(
            "thread-id", TRANSIENT_THREAD_ID,
            "task-id", TASK_ID_1,
            "processor-node-id", PROCESSOR_NODE_ID,
            "topic", TOPIC_NAME))
    );

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

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

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

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