@Override
  public void metricChange(final KafkaMetric metric) {
    if (!metric.metricName().name().equals("total-sst-files-size")) {
      return;
    }

    handleNewSstFilesSizeMetric(
        metric,
        metric.metricName().tags().getOrDefault(TASK_ID_TAG, ""),
        getQueryId(metric)
    );
  }
@Test
  public void shouldIgnoreNonSSTMetrics() {
    // When:
    listener.metricChange(mockMetric(
      KAFKA_METRIC_GROUP,
      "other-metric",
      BigInteger.valueOf(2),
      ImmutableMap.of("store-id", "s1", "task-id", "t1", "thread-id", THREAD_ID)));

    // Then:
    assertThrows(AssertionError.class, () -> verifyAndGetRegisteredMetric(TASK_STORAGE_METRIC, TASK_ONE_TAGS));
  }