private void metricChange(KafkaMetric metric) {
            MetricKey metricKey = metricNamingStrategy.metricKey(metric.metricName());
            metricMap.put(metricKey, metric);
            if (doubleDeltas.contains(metricKey)) {
                log.warn("Registering a new metric {} which already has a last value tracked. " +
                    "Removing metric from delta register.", metric.metricName(), new Exception());

                /*
                 This scenario shouldn't occur while registering a metric since it should
                 have already been cleared out on cleanup/shutdown.
                 We remove the metric here to clear out the delta register because we are running
                 into an issue where old metrics are being re-registered which causes us to
                 record a negative delta
                */
                doubleDeltas.remove(metricKey);
            }
            metricAdded.put(metricKey, Instant.ofEpochMilli(time.milliseconds()));
        }
Keeps track of the state of metrics, e.g. when they were added, what their getAndSet value is,
and clearing them out when they're removed.
clients/src/main/java/org/apache/kafka/common/telemetry/internals/KafkaMetricsCollector.java