@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;
    }
@Test
    public void testClientInstanceId() {
        try (AdminClientUnitTestEnv env = mockClientEnv()) {
            Uuid expected = Uuid.randomUuid();

            GetTelemetrySubscriptionsResponseData responseData =
                new GetTelemetrySubscriptionsResponseData().setClientInstanceId(expected).setErrorCode(Errors.NONE.code());

            env.kafkaClient().prepareResponse(
                request -> request instanceof GetTelemetrySubscriptionsRequest,
                new GetTelemetrySubscriptionsResponse(responseData));

            Uuid result = env.adminClient().clientInstanceId(Duration.ofSeconds(1));
            assertEquals(expected, result);
        }
    }