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

            final ClientTelemetryState oldState;
            final ClientTelemetrySubscription oldSubscription;
            lock.readLock().lock();
            try {
                oldState = state;
                oldSubscription = subscription;
            } finally {
                lock.readLock().unlock();
            }

            Optional<Integer> errorIntervalMsOpt = ClientTelemetryUtils.maybeFetchErrorIntervalMs(data.errorCode(),
                oldSubscription != null ? oldSubscription.pushIntervalMs() : -1);
            /*
             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 retried.
            */
            if (errorIntervalMsOpt.isPresent()) {
                /*
                 Update the state from SUBSCRIPTION_INR_PROGRESS to SUBSCRIPTION_NEEDED as the error
                 response indicates that the subscription is not valid.
                */
                if (!maybeSetState(ClientTelemetryState.SUBSCRIPTION_NEEDED)) {
                    log.warn("Unable to transition state after failed get telemetry subscriptions from state {}", oldState);
                }
                updateErrorResult(errorIntervalMsOpt.get(), nowMs);
                return;
            }

            Uuid clientInstanceId = ClientTelemetryUtils.validateClientInstanceId(data.clientInstanceId());
            int intervalMs = ClientTelemetryUtils.validateIntervalMs(data.pushIntervalMs());
            Predicate<? super MetricKeyable> selector = ClientTelemetryUtils.getSelectorFromRequestedMetrics(
                data.requestedMetrics());
            List<CompressionType> acceptedCompressionTypes = ClientTelemetryUtils.getCompressionTypesFromAcceptedList(
                data.acceptedCompressionTypes());

            /*
             Check if the delta temporality has changed, if so, we need to reset the ledger tracking
             the last value sent for each metric.
            */
            if (oldSubscription != null && oldSubscription.deltaTemporality() != data.deltaTemporality()) {
                log.info("Delta temporality has changed from {} to {}, resetting metric values",
                    oldSubscription.deltaTemporality(), data.deltaTemporality());
                if (kafkaMetricsCollector != null) {
                    kafkaMetricsCollector.metricsReset();
                }
            }

            ClientTelemetrySubscription clientTelemetrySubscription = new ClientTelemetrySubscription(
                clientInstanceId,
                data.subscriptionId(),
                intervalMs,
                acceptedCompressionTypes,
                data.deltaTemporality(),
                selector);

            lock.writeLock().lock();
            try {
                /*
                 This is the case if we began termination sometime after the subscription request
                 was issued. We're just now getting our callback, but we need to ignore it.
                */
                if (isTerminatingState()) {
                    return;
                }

                ClientTelemetryState newState;
                if (selector == ClientTelemetryUtils.SELECTOR_NO_METRICS) {
                    /*
                     This is the case where no metrics are requested and/or match the filters. We need
                     to wait intervalMs then retry.
                    */
                    newState = ClientTelemetryState.SUBSCRIPTION_NEEDED;
                } else {
                    newState = ClientTelemetryState.PUSH_NEEDED;
                }

                // If we're terminating, don't update state or set the subscription.
                if (!maybeSetState(newState)) {
                    return;
                }

                updateSubscriptionResult(clientTelemetrySubscription, nowMs);
                log.info("Client telemetry registered with client instance id: {}", subscription.clientInstanceId());
            } finally {
                lock.writeLock().unlock();
            }
        }
Method is invoked whenever a metric is removed
clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java