public synchronized ClientInstanceIds clientInstanceIds(final Duration timeout) {
        if (timeout.isNegative()) {
            throw new IllegalArgumentException("The timeout cannot be negative.");
        }
        if (state().hasNotStarted()) {
            throw new IllegalStateException("KafkaStreams has not been started, you can retry after calling start().");
        }
        if (state().isShuttingDown() || state.hasCompletedShutdown()) {
            throw new IllegalStateException("KafkaStreams has been stopped (" + state + ").");
        }

        final Timer remainingTime = time.timer(timeout.toMillis());
        final ClientInstanceIdsImpl clientInstanceIds = new ClientInstanceIdsImpl();

        // (1) fan-out calls to threads

        // StreamThread for main/restore consumers and producer(s)
        final Map<String, KafkaFuture<Uuid>> consumerFutures = new HashMap<>();
        final Map<String, KafkaFuture<Map<String, KafkaFuture<Uuid>>>> producerFutures = new HashMap<>();
        synchronized (changeThreadCount) {
            for (final StreamThread streamThread : threads) {
                consumerFutures.putAll(streamThread.consumerClientInstanceIds(timeout));
                producerFutures.put(streamThread.getName(), streamThread.producersClientInstanceIds(timeout));
            }
        }

        // GlobalThread
        KafkaFuture<Uuid> globalThreadFuture = null;
        if (globalStreamThread != null) {
            globalThreadFuture = globalStreamThread.globalConsumerInstanceId(timeout);
        }

        // (2) get admin client instance id in a blocking fashion, while Stream/GlobalThreads work in parallel
        try {
            clientInstanceIds.setAdminInstanceId(adminClient.clientInstanceId(timeout));
            remainingTime.update(time.milliseconds());
        } catch (final IllegalStateException telemetryDisabledError) {
            // swallow
            log.debug("Telemetry is disabled on the admin client.");
        } catch (final TimeoutException timeoutException) {
            throw timeoutException;
        } catch (final Exception error) {
            throw new StreamsException("Could not retrieve admin client instance id.", error);
        }

        // (3) collect client instance ids from threads

        // (3a) collect consumers from StreamsThread
        for (final Map.Entry<String, KafkaFuture<Uuid>> consumerFuture : consumerFutures.entrySet()) {
            final Uuid instanceId = getOrThrowException(
                consumerFuture.getValue(),
                remainingTime.remainingMs(),
                () -> String.format(
                    "Could not retrieve consumer instance id for %s.",
                    consumerFuture.getKey()
                )
            );
            remainingTime.update(time.milliseconds());

            // could be `null` if telemetry is disabled on the consumer itself
            if (instanceId != null) {
                clientInstanceIds.addConsumerInstanceId(
                    consumerFuture.getKey(),
                    instanceId
                );
            } else {
                log.debug(String.format("Telemetry is disabled for %s.", consumerFuture.getKey()));
            }
        }

        // (3b) collect producers from StreamsThread
        for (final Map.Entry<String, KafkaFuture<Map<String, KafkaFuture<Uuid>>>> threadProducerFuture : producerFutures.entrySet()) {
            final Map<String, KafkaFuture<Uuid>> streamThreadProducerFutures = getOrThrowException(
                threadProducerFuture.getValue(),
                remainingTime.remainingMs(),
                () -> String.format(
                    "Could not retrieve producer instance id for %s.",
                    threadProducerFuture.getKey()
                )
            );
            remainingTime.update(time.milliseconds());

            for (final Map.Entry<String, KafkaFuture<Uuid>> producerFuture : streamThreadProducerFutures.entrySet()) {
                final Uuid instanceId = getOrThrowException(
                    producerFuture.getValue(),
                    remainingTime.remainingMs(),
                    () -> String.format(
                        "Could not retrieve producer instance id for %s.",
                        producerFuture.getKey()
                    )
                );
                remainingTime.update(time.milliseconds());

                // could be `null` if telemetry is disabled on the producer itself
                if (instanceId != null) {
                    clientInstanceIds.addProducerInstanceId(
                        producerFuture.getKey(),
                        instanceId
                    );
                } else {
                    log.debug(String.format("Telemetry is disabled for %s.", producerFuture.getKey()));
                }
            }
        }

        // (3c) collect from GlobalThread
        if (globalThreadFuture != null) {
            final Uuid instanceId = getOrThrowException(
                globalThreadFuture,
                remainingTime.remainingMs(),
                () -> "Could not retrieve global consumer client instance id."
            );
            remainingTime.update(time.milliseconds());

            // could be `null` if telemetry is disabled on the client itself
            if (instanceId != null) {
                clientInstanceIds.addConsumerInstanceId(
                    globalStreamThread.getName(),
                    instanceId
                );
            } else {
                log.debug("Telemetry is disabled for the global consumer.");
            }
        }

        return clientInstanceIds;
    }
@Test
    public void shouldThrowOnClientInstanceIdsWhenNotStarted() {
        prepareStreams();
        prepareStreamThread(streamThreadOne, 1);
        prepareStreamThread(streamThreadTwo, 2);

        try (final KafkaStreams streams = new KafkaStreams(getBuilderWithSource().build(), props, supplier, time)) {
            final IllegalStateException error = assertThrows(
                IllegalStateException.class,
                () -> streams.clientInstanceIds(Duration.ZERO)
            );
            assertThat(
                error.getMessage(),
                equalTo("KafkaStreams has not been started, you can retry after calling start().")
            );
        }
    }