@SuppressWarnings("deprecation") // Needed to include StreamsConfig.EXACTLY_ONCE_BETA in error log for UnsupportedVersionException
    boolean runLoop() {
        subscribeConsumer();

        // if the thread is still in the middle of a rebalance, we should keep polling
        // until the rebalance is completed before we close and commit the tasks
        while (isRunning() || taskManager.rebalanceInProgress()) {
            try {
                checkForTopologyUpdates();
                // If we received the shutdown signal while waiting for a topology to be added, we can
                // stop polling regardless of the rebalance status since we know there are no tasks left
                if (!isRunning() && topologyMetadata.isEmpty()) {
                    log.info("Shutting down thread with empty topology.");
                    break;
                }

                maybeSendShutdown();
                final long size = cacheResizeSize.getAndSet(-1L);
                if (size != -1L) {
                    cacheResizer.accept(size);
                }
                if (processingThreadsEnabled) {
                    runOnceWithProcessingThreads();
                } else {
                    runOnceWithoutProcessingThreads();
                }

                maybeGetClientInstanceIds();

                // Check for a scheduled rebalance but don't trigger it until the current rebalance is done
                if (!taskManager.rebalanceInProgress() && nextProbingRebalanceMs.get() < time.milliseconds()) {
                    log.info("Triggering the followup rebalance scheduled for {}.", Utils.toLogDateTimeFormat(nextProbingRebalanceMs.get()));
                    mainConsumer.enforceRebalance("triggered followup rebalance scheduled for " + nextProbingRebalanceMs.get());
                    nextProbingRebalanceMs.set(Long.MAX_VALUE);
                }
            } catch (final TaskCorruptedException e) {
                log.warn("Detected the states of tasks " + e.corruptedTasks() + " are corrupted. " +
                         "Will close the task as dirty and re-create and bootstrap from scratch.", e);
                try {
                    // check if any active task got corrupted. We will trigger a rebalance in that case.
                    // once the task corruptions have been handled
                    final boolean enforceRebalance = taskManager.handleCorruption(e.corruptedTasks());
                    if (enforceRebalance && eosEnabled) {
                        log.info("Active task(s) got corrupted. Triggering a rebalance.");
                        mainConsumer.enforceRebalance("Active tasks corrupted");
                    }
                } catch (final TaskMigratedException taskMigrated) {
                    handleTaskMigrated(taskMigrated);
                }
            } catch (final TaskMigratedException e) {
                handleTaskMigrated(e);
            } catch (final UnsupportedVersionException e) {
                final String errorMessage = e.getMessage();
                if (errorMessage != null &&
                    errorMessage.startsWith("Broker unexpectedly doesn't support requireStable flag on version ")) {

                    log.error("Shutting down because the Kafka cluster seems to be on a too old version. " +
                              "Setting {}=\"{}\"/\"{}\" requires broker version 2.5 or higher.",
                          StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
                          StreamsConfig.EXACTLY_ONCE_V2, StreamsConfig.EXACTLY_ONCE_BETA);
                }
                failedStreamThreadSensor.record();
                this.streamsUncaughtExceptionHandler.accept(new StreamsException(e), false);
                return false;
            } catch (final StreamsException e) {
                throw e;
            } catch (final Exception e) {
                throw new StreamsException(e);
            }
        }
        return true;
    }
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