void maybeGetClientInstanceIds() {
        // we pass in a timeout of zero into each `clientInstanceId()` call
        // to just trigger the "get instance id" background RPC;
        // we don't want to block the stream thread that can do useful work in the meantime

        if (fetchDeadlineClientInstanceId != -1) {
            if (!mainConsumerInstanceIdFuture.isDone()) {
                if (fetchDeadlineClientInstanceId >= time.milliseconds()) {
                    try {
                        mainConsumerInstanceIdFuture.complete(mainConsumer.clientInstanceId(Duration.ZERO));
                    } catch (final IllegalStateException disabledError) {
                        // if telemetry is disabled on a client, we swallow the error,
                        // to allow returning a partial result for all other clients
                        mainConsumerInstanceIdFuture.complete(null);
                    } catch (final TimeoutException swallow) {
                        // swallow
                    } catch (final Exception error) {
                        mainConsumerInstanceIdFuture.completeExceptionally(error);
                    }
                } else {
                    mainConsumerInstanceIdFuture.completeExceptionally(
                        new TimeoutException("Could not retrieve main consumer client instance id.")
                    );
                }
            }


            if (!stateUpdaterEnabled && !restoreConsumerInstanceIdFuture.isDone()) {
                if (fetchDeadlineClientInstanceId >= time.milliseconds()) {
                    try {
                        restoreConsumerInstanceIdFuture.complete(restoreConsumer.clientInstanceId(Duration.ZERO));
                    } catch (final IllegalStateException disabledError) {
                        // if telemetry is disabled on a client, we swallow the error,
                        // to allow returning a partial result for all other clients
                        restoreConsumerInstanceIdFuture.complete(null);
                    } catch (final TimeoutException swallow) {
                        // swallow
                    } catch (final Exception error) {
                        restoreConsumerInstanceIdFuture.completeExceptionally(error);
                    }
                } else {
                    restoreConsumerInstanceIdFuture.completeExceptionally(
                        new TimeoutException("Could not retrieve restore consumer client instance id.")
                    );
                }
            }

            if (!processingMode.equals(StreamsConfigUtils.ProcessingMode.EXACTLY_ONCE_ALPHA) &&
                    !threadProducerInstanceIdFuture.isDone()) {

                if (fetchDeadlineClientInstanceId >= time.milliseconds()) {
                    try {
                        threadProducerInstanceIdFuture.complete(
                            taskManager.threadProducer().kafkaProducer().clientInstanceId(Duration.ZERO)
                        );
                    } catch (final IllegalStateException disabledError) {
                        // if telemetry is disabled on a client, we swallow the error,
                        // to allow returning a partial result for all other clients
                        threadProducerInstanceIdFuture.complete(null);
                    } catch (final TimeoutException swallow) {
                        // swallow
                    } catch (final Exception error) {
                        threadProducerInstanceIdFuture.completeExceptionally(error);
                    }
                } else {
                    threadProducerInstanceIdFuture.completeExceptionally(
                        new TimeoutException("Could not retrieve thread producer client instance id.")
                    );
                }
            }

            maybeResetFetchDeadline();
        }
    }
Main event loop for polling, and processing records through topologies.

@throws IllegalStateException If store gets registered after initialized is already finished
@throws StreamsException      if the store's change log does not contain the partition
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java