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

    // Then:
    final Gauge<?> queryGauge = verifyAndGetRegisteredMetric(QUERY_STORAGE_METRIC, QUERY_TAGS);
    final Object queryValue = queryGauge.value(null, 0);
    final Gauge<?> taskGaugeOne = verifyAndGetRegisteredMetric(TASK_STORAGE_METRIC, TASK_ONE_TAGS);
    final Object taskValueOne = taskGaugeOne.value(null, 0);
    final Gauge<?> taskGaugeTwo = verifyAndGetRegisteredMetric(TASK_STORAGE_METRIC, TASK_TWO_TAGS);
    final Object taskValueTwo = taskGaugeTwo.value(null, 0);


    assertThat(taskValueOne, equalTo(BigInteger.valueOf(2)));
    assertThat(taskValueTwo, equalTo(BigInteger.valueOf(5)));
    assertThat(queryValue, equalTo(BigInteger.valueOf(7)));
  }