private ClientMetricsInstance createClientInstance(Uuid clientInstanceId, ClientMetricsInstanceMetadata instanceMetadata) {

        int pushIntervalMs = ClientMetricsConfigs.DEFAULT_INTERVAL_MS;
        // Keep a set of metrics to avoid duplicates in case of overlapping subscriptions.
        Set<String> subscribedMetrics = new HashSet<>();
        boolean allMetricsSubscribed = false;

        int currentSubscriptionVersion = subscriptionUpdateVersion.get();
        for (SubscriptionInfo info : subscriptionMap.values()) {
            if (instanceMetadata.isMatch(info.matchPattern())) {
                allMetricsSubscribed = allMetricsSubscribed || info.metrics().contains(
                    ClientMetricsConfigs.ALL_SUBSCRIBED_METRICS_CONFIG);
                subscribedMetrics.addAll(info.metrics());
                pushIntervalMs = Math.min(pushIntervalMs, info.intervalMs());
            }
        }

        /*
         If client matches with any subscription that has * metrics string, then it means that client
         is subscribed to all the metrics, so just send the * string as the subscribed metrics.
        */
        if (allMetricsSubscribed) {
            // Only add an * to indicate that all metrics are subscribed.
            subscribedMetrics.clear();
            subscribedMetrics.add(ClientMetricsConfigs.ALL_SUBSCRIBED_METRICS_CONFIG);
        }

        int subscriptionId = computeSubscriptionId(subscribedMetrics, pushIntervalMs, clientInstanceId);

        return new ClientMetricsInstance(clientInstanceId, instanceMetadata, subscriptionId,
            currentSubscriptionVersion, subscribedMetrics, pushIntervalMs);
    }
Handles client telemetry metrics requests/responses, subscriptions and instance information.
server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java