private ClientMetricsInstance clientInstance(Uuid clientInstanceId, RequestContext requestContext) {
        ClientMetricsInstance clientInstance = clientInstanceCache.get(clientInstanceId);

        if (clientInstance == null) {
            /*
             If the client instance is not present in the cache, then create a new client instance
             and update the cache. This can also happen when the telemetry request is received by
             the separate broker instance.
             Though cache is synchronized, but it is possible that concurrent calls can create the same
             client instance. Hence, safeguard the client instance creation with a double-checked lock
             to ensure that only one instance is created.
            */
            synchronized (this) {
                clientInstance = clientInstanceCache.get(clientInstanceId);
                if (clientInstance != null) {
                    return clientInstance;
                }

                ClientMetricsInstanceMetadata instanceMetadata = new ClientMetricsInstanceMetadata(
                    clientInstanceId, requestContext);
                clientInstance = createClientInstanceAndUpdateCache(clientInstanceId, instanceMetadata);
            }
        } else if (clientInstance.subscriptionVersion() < subscriptionUpdateVersion.get()) {
            /*
             If the last subscription update version for client instance is older than the subscription
             updated version, then re-evaluate the subscription information for the client as per the
             updated subscriptions. This is to ensure that the client instance is always in sync with
             the latest subscription information.
             Though cache is synchronized, but it is possible that concurrent calls can create the same
             client instance. Hence, safeguard the client instance update with a double-checked lock
              to ensure that only one instance is created.
            */
            synchronized (this) {
                clientInstance = clientInstanceCache.get(clientInstanceId);
                if (clientInstance.subscriptionVersion() >= subscriptionUpdateVersion.get()) {
                    return clientInstance;
                }
                // Cancel the existing expiration timer task for the old client instance.
                clientInstance.cancelExpirationTimerTask();
                clientInstance = createClientInstanceAndUpdateCache(clientInstanceId, clientInstance.instanceMetadata());
            }
        }

        // Update the expiration timer task for the client instance.
        long expirationTimeMs = Math.max(cacheExpiryMs, clientInstance.pushIntervalMs() * 3);
        TimerTask timerTask = new ExpirationTimerTask(clientInstanceId, expirationTimeMs);
        clientInstance.updateExpirationTimerTask(timerTask);
        expirationTimer.add(timerTask);

        return clientInstance;
    }
Handles client telemetry metrics requests/responses, subscriptions and instance information.
server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java