public GetTelemetrySubscriptionsResponse processGetTelemetrySubscriptionRequest(
        GetTelemetrySubscriptionsRequest request, RequestContext requestContext) {

        long now = time.milliseconds();
        Uuid clientInstanceId = Optional.ofNullable(request.data().clientInstanceId())
            .filter(id -> !id.equals(Uuid.ZERO_UUID))
            .orElse(generateNewClientId());

        /*
         Get the client instance from the cache or create a new one. If subscription has changed
         since the last request, then the client instance will be re-evaluated. Validation of the
         request will be done after the client instance is created. If client issues another get
         telemetry request prior to push interval, then the client should get a throttle error but if
         the subscription has changed since the last request then the client should get the updated
         subscription immediately.
        */
        ClientMetricsInstance clientInstance = clientInstance(clientInstanceId, requestContext);

        try {
            // Validate the get request parameters for the client instance.
            validateGetRequest(request, clientInstance, now);
        } catch (ApiException exception) {
            return request.getErrorResponse(0, exception);
        }

        clientInstance.lastKnownError(Errors.NONE);
        return createGetSubscriptionResponse(clientInstanceId, clientInstance);
    }
Handles client telemetry metrics requests/responses, subscriptions and instance information.
server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java