@Override
    public Uuid clientInstanceId(Duration timeout) {
        if (timeout.isNegative()) {
            throw new IllegalArgumentException("The timeout cannot be negative.");
        }

        if (!clientTelemetryEnabled) {
            throw new IllegalStateException("Telemetry is not enabled. Set config `" + AdminClientConfig.ENABLE_METRICS_PUSH_CONFIG + "` to `true`.");
        }

        if (clientInstanceId != null) {
            return clientInstanceId;
        }

        final long now = time.milliseconds();
        final KafkaFutureImpl<Uuid> future = new KafkaFutureImpl<>();
        runnable.call(new Call("getTelemetrySubscriptions", calcDeadlineMs(now, (int) timeout.toMillis()),
            new LeastLoadedNodeProvider()) {

            @Override
            GetTelemetrySubscriptionsRequest.Builder createRequest(int timeoutMs) {
                return new GetTelemetrySubscriptionsRequest.Builder(new GetTelemetrySubscriptionsRequestData(), true);
            }

            @Override
            void handleResponse(AbstractResponse abstractResponse) {
                GetTelemetrySubscriptionsResponse response = (GetTelemetrySubscriptionsResponse) abstractResponse;
                if (response.error() != Errors.NONE) {
                    future.completeExceptionally(response.error().exception());
                } else {
                    future.complete(response.data().clientInstanceId());
                }
            }

            @Override
            void handleFailure(Throwable throwable) {
                future.completeExceptionally(throwable);
            }
        }, now);

        try {
            clientInstanceId = future.get();
        } catch (Exception e) {
            log.error("Error occurred while fetching client instance id", e);
            throw new KafkaException("Error occurred while fetching client instance id", e);
        }

        return clientInstanceId;
    }
Returns the broker id pertaining to the given resource, or null if the resource is not associated
with a particular broker.
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java