@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 shouldThrowWhenTopicNameTagIsMissing() {
    // When:
    assertThrows(
      KsqlException.class,
      () -> listener.metricChange(mockMetric(
        BYTES_CONSUMED_TOTAL,
        2D,
        ImmutableMap.of(
          "thread-id", THREAD_ID,
          "task-id", TASK_ID_1,
          "processor-node-id", PROCESSOR_NODE_ID))
      )
    );
  }