@Override
        public void handleResponse(PushTelemetryResponse response) {
            final long nowMs = time.milliseconds();
            final PushTelemetryResponseData data = response.data();

            lock.writeLock().lock();
            try {
                /*
                 This is the case when client began termination sometime after the last push request
                 was issued. Just getting the callback, hence need to ignore it.
                */
                if (isTerminatingState()) {
                    return;
                }

                Optional<Integer> errorIntervalMsOpt = ClientTelemetryUtils.maybeFetchErrorIntervalMs(data.errorCode(),
                    subscription.pushIntervalMs());
                /*
                 If the error code indicates that the interval ms needs to be updated as per the error
                 code then update the interval ms and state so that the subscription can be re-fetched,
                 and the push retried.
                */
                if (errorIntervalMsOpt.isPresent()) {
                    if (!maybeSetState(ClientTelemetryState.SUBSCRIPTION_NEEDED)) {
                        log.warn("Unable to transition state after failed push telemetry from state {}", state);
                    }
                    updateErrorResult(errorIntervalMsOpt.get(), nowMs);
                    return;
                }

                lastRequestMs = nowMs;
                intervalMs = subscription.pushIntervalMs();
                if (!maybeSetState(ClientTelemetryState.PUSH_NEEDED)) {
                    log.warn("Unable to transition state after successful push telemetry from state {}", state);
                }
            } finally {
                lock.writeLock().unlock();
            }
        }
Method is invoked whenever a metric is removed
clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java