public synchronized KafkaMetric removeMetric(MetricName metricName) {
        KafkaMetric metric = this.metrics.remove(metricName);
        if (metric != null) {
            for (MetricsReporter reporter : reporters) {
                try {
                    reporter.metricRemoval(metric);
                } catch (Exception e) {
                    log.error("Error when removing metric from " + reporter.getClass().getName(), e);
                }
            }
            log.trace("Removed metric named {}", metricName);
        }
        return metric;
    }
@Test
    public void testRemoveMetric() {
        int size = metrics.metrics().size();
        metrics.addMetric(metrics.metricName("test1", "grp1"), new WindowedCount());
        metrics.addMetric(metrics.metricName("test2", "grp1"), new WindowedCount());

        assertNotNull(metrics.removeMetric(metrics.metricName("test1", "grp1")));
        assertNull(metrics.metrics().get(metrics.metricName("test1", "grp1")));
        assertNotNull(metrics.metrics().get(metrics.metricName("test2", "grp1")));

        assertNotNull(metrics.removeMetric(metrics.metricName("test2", "grp1")));
        assertNull(metrics.metrics().get(metrics.metricName("test2", "grp1")));

        assertEquals(size, metrics.metrics().size());
    }