@Override
  public void metricChange(final KafkaMetric metric) {
    if (!metric.metricName().group().equals(StreamsMetricsImpl.STATE_STORE_LEVEL_GROUP)) {
      return;
    }
    metricRemoval(metric);
    final Collection<AggregatedMetric<?>> registered
        = registeredMetrics.get(metric.metricName().name());
    if (registered == null) {
      return;
    }
    registered.forEach(r -> r.add(metric));
  }
@Test
  public void shouldIgnoreMetricsFromWrongGroup() {
    // When:
    collector.metricChange(mockMetric(
        "some-group",
        RocksDBMetricsCollector.BLOCK_CACHE_USAGE,
        "a",
        BigInteger.valueOf(123)
    ));

    // Then:
    final Gauge<?> value =
        verifyAndGetRegisteredMetric(RocksDBMetricsCollector.BLOCK_CACHE_USAGE + "-total");
    assertThat(value.value(null, 0), equalTo(BigInteger.valueOf(0)));
  }