@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 shouldAddNewGauges() {
    // Given:
    listener.metricChange(mockMetric(
      KAFKA_METRIC_GROUP,
      KAFKA_METRIC_NAME,
      BigInteger.valueOf(2),
      ImmutableMap.of("task-id", "t1", "thread-id", THREAD_ID))
    );

    // When:

    final Gauge<?> queryGauge = verifyAndGetRegisteredMetric(QUERY_STORAGE_METRIC, QUERY_TAGS);
    final Object queryValue = queryGauge.value(null, 0);
    final Gauge<?> taskGauge = verifyAndGetRegisteredMetric(TASK_STORAGE_METRIC, TASK_ONE_TAGS);
    final Object taskValue = taskGauge.value(null, 0);

    // Then:
    assertThat(taskValue, equalTo(BigInteger.valueOf(2)));
    assertThat(queryValue, equalTo(BigInteger.valueOf(2)));

  }