@Override
        public void initiateClose(long timeoutMs) {
            log.debug("initiate close for client telemetry, check if terminal push required. Timeout {} ms.", timeoutMs);

            lock.writeLock().lock();
            try {
                // If we never fetched a subscription, we can't really push anything.
                if (lastRequestMs == 0) {
                    log.debug("Telemetry subscription not loaded, not attempting terminating push");
                    return;
                }

                if (state == ClientTelemetryState.SUBSCRIPTION_NEEDED) {
                    log.debug("Subscription not yet loaded, ignoring terminal push");
                    return;
                }

                if (isTerminatingState() || !maybeSetState(ClientTelemetryState.TERMINATING_PUSH_NEEDED)) {
                    log.debug("Ignoring subsequent initiateClose");
                    return;
                }

                try {
                    log.info("About to wait {} ms. for terminal telemetry push to be submitted", timeoutMs);
                    if (!terminalPushInProgress.await(timeoutMs, TimeUnit.MILLISECONDS)) {
                        log.info("Wait for terminal telemetry push to be submitted has elapsed; may not have actually sent request");
                    }
                } catch (InterruptedException e) {
                    log.warn("Error during client telemetry close", e);
                }
            } finally {
                lock.writeLock().unlock();
            }
        }
Method is invoked whenever a metric is removed
clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java