public void increment(K key) {
        synchronized (this) {
            if (closed) {
                log.warn("{}: Attempted to increment {}, but the GaugeSuite was closed.",
                    suiteName, key.toString());
                return;
            }
            StoredIntGauge gauge = gauges.get(key);
            if (gauge != null) {
                // Fast path: increment the existing counter.
                if (gauge.increment() > 0) {
                    removable.remove(key);
                }
                return;
            }
            if (gauges.size() == maxEntries) {
                if (removable.isEmpty()) {
                    log.debug("{}: Attempted to increment {}, but there are already {} entries.",
                        suiteName, key.toString(), maxEntries);
                    return;
                }
                Iterator<K> iter = removable.iterator();
                K keyToRemove = iter.next();
                iter.remove();
                MetricName metricNameToRemove = gauges.get(keyToRemove).metricName;
                gauges.remove(keyToRemove);
                pending.push(new PendingMetricsChange(metricNameToRemove, null));
                log.trace("{}: Removing the metric {}, which has a value of 0.",
                    suiteName, keyToRemove.toString());
            }
            MetricName metricNameToAdd = metricNameCalculator.apply(key);
            gauge = new StoredIntGauge(metricNameToAdd);
            gauges.put(key, gauge);
            pending.push(new PendingMetricsChange(metricNameToAdd, gauge));
            log.trace("{}: Adding a new metric {}.", suiteName, key.toString());
        }
        // Drop the object monitor and perform any pending metrics additions or removals.
        performPendingMetricsOperations();
    }
This callback is invoked when the metrics system retrieves the value of this gauge.
clients/src/main/java/org/apache/kafka/common/metrics/internals/IntGaugeSuite.java