@Override
        public long timeToNextUpdate(long requestTimeoutMs) {
            final long nowMs = time.milliseconds();
            final ClientTelemetryState localState;
            final long localLastRequestMs;
            final int localIntervalMs;

            lock.readLock().lock();
            try {
                if (!enabled) {
                    return Integer.MAX_VALUE;
                }

                localState = state;
                localLastRequestMs = lastRequestMs;
                localIntervalMs = intervalMs;
            } finally {
                lock.readLock().unlock();
            }

            final long timeMs;
            final String apiName;
            final String msg;
            final boolean isTraceEnabled = log.isTraceEnabled();

            switch (localState) {
                case SUBSCRIPTION_IN_PROGRESS:
                case PUSH_IN_PROGRESS:
                    /*
                     We have a network request in progress. We record the time of the request
                     submission, so wait that amount of the time PLUS the requestTimeout that
                     is provided.
                    */
                    apiName = (localState == ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS) ? ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS.name : ApiKeys.PUSH_TELEMETRY.name;
                    timeMs = requestTimeoutMs;
                    msg = isTraceEnabled ? "" : String.format("the remaining wait time for the %s network API request, as specified by %s", apiName, CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG);
                    break;
                case TERMINATING_PUSH_IN_PROGRESS:
                    timeMs = Long.MAX_VALUE;
                    msg = isTraceEnabled ? "" : "the terminating push is in progress, disabling telemetry for further requests";
                    break;
                case TERMINATING_PUSH_NEEDED:
                    timeMs = 0;
                    msg = isTraceEnabled ? "" : String.format("the client should try to submit the final %s network API request ASAP before closing", ApiKeys.PUSH_TELEMETRY.name);
                    break;
                case SUBSCRIPTION_NEEDED:
                case PUSH_NEEDED:
                    apiName = (localState == ClientTelemetryState.SUBSCRIPTION_NEEDED) ? ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS.name : ApiKeys.PUSH_TELEMETRY.name;
                    long timeRemainingBeforeRequest = localLastRequestMs + localIntervalMs - nowMs;
                    if (timeRemainingBeforeRequest <= 0) {
                        timeMs = 0;
                        msg = isTraceEnabled ? "" : String.format("the wait time before submitting the next %s network API request has elapsed", apiName);
                    } else {
                        timeMs = timeRemainingBeforeRequest;
                        msg = isTraceEnabled ? "" : String.format("the client will wait before submitting the next %s network API request", apiName);
                    }
                    break;
                default:
                    throw new IllegalStateException("Unknown telemetry state: " + localState);
            }

            if (isTraceEnabled) {
                log.trace("For telemetry state {}, returning the value {} ms; {}", localState, timeMs, msg);
            }
            return timeMs;
        }
Method is invoked whenever a metric is removed
clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java