private Optional<Builder<?>> createSubscriptionRequest(ClientTelemetrySubscription localSubscription) {
            /*
             If we've previously retrieved a subscription, it will contain the client instance ID
             that the broker assigned. Otherwise, per KIP-714, we send a special "null" UUID to
             signal to the broker that we need to have a client instance ID assigned.
            */
            Uuid clientInstanceId = (localSubscription != null) ? localSubscription.clientInstanceId() : Uuid.ZERO_UUID;
            log.debug("Creating telemetry subscription request with client instance id {}", clientInstanceId);

            lock.writeLock().lock();
            try {
                if (isTerminatingState()) {
                    return Optional.empty();
                }

                if (!maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS)) {
                    return Optional.empty();
                }
            } finally {
                lock.writeLock().unlock();
            }

            AbstractRequest.Builder<?> requestBuilder = new GetTelemetrySubscriptionsRequest.Builder(
                new GetTelemetrySubscriptionsRequestData().setClientInstanceId(clientInstanceId), true);
            return Optional.of(requestBuilder);
        }
Method is invoked whenever a metric is removed
clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java