public void updateSubscription(String subscriptionName, Properties properties) {
        // Validate the subscription properties.
        ClientMetricsConfigs.validate(subscriptionName, properties);
        // IncrementalAlterConfigs API will send empty configs when all the configs are deleted
        // for respective subscription. In that case, we need to remove the subscription from the map.
        if (properties.isEmpty()) {
            // Remove the subscription from the map if it exists, else ignore the config update.
            if (subscriptionMap.containsKey(subscriptionName)) {
                log.info("Removing subscription [{}] from the subscription map", subscriptionName);
                subscriptionMap.remove(subscriptionName);
                subscriptionUpdateVersion.incrementAndGet();
            }
            return;
        }

        updateClientSubscription(subscriptionName, new ClientMetricsConfigs(properties));
        /*
         Increment subscription update version to indicate that there is a change in the subscription. This will
         be used to determine if the next telemetry request needs to re-evaluate the subscription id
         as per the changed subscriptions.
        */
        subscriptionUpdateVersion.incrementAndGet();
    }
Handles client telemetry metrics requests/responses, subscriptions and instance information.
server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java