@Override
    public void close() {
        removeMetric(FLUSH);
        removeMetric(TXN_INIT);
        removeMetric(TXN_BEGIN);
        removeMetric(TXN_SEND_OFFSETS);
        removeMetric(TXN_COMMIT);
        removeMetric(TXN_ABORT);
        removeMetric(METADATA_WAIT);
    }
@Test
    public void shouldRemoveMetricsOnClose() {
        // When:
        producerMetrics.close();

        // Then:
        assertMetricRemoved(FLUSH_TIME_TOTAL);
        assertMetricRemoved(TXN_INIT_TIME_TOTAL);
        assertMetricRemoved(TXN_BEGIN_TIME_TOTAL);
        assertMetricRemoved(TXN_COMMIT_TIME_TOTAL);
        assertMetricRemoved(TXN_ABORT_TIME_TOTAL);
        assertMetricRemoved(TXN_SEND_OFFSETS_TIME_TOTAL);
        assertMetricRemoved(METADATA_WAIT_TIME_TOTAL);
    }