@Override
    public void run() {
        log.info("Starting");
        if (setState(State.STARTING) == null) {
            log.info("StreamThread already shutdown. Not running");
            return;
        }
        boolean cleanRun = false;
        try {
            cleanRun = runLoop();
        } catch (final Throwable e) {
            failedStreamThreadSensor.record();
            requestLeaveGroupDuringShutdown();
            streamsUncaughtExceptionHandler.accept(e, false);
            // Note: the above call currently rethrows the exception, so nothing below this line will be executed
        } finally {
            completeShutdown(cleanRun);
        }
    }
Execute the stream processors

@throws KafkaException   for any Kafka-related exceptions
@throws RuntimeException for any other non-Kafka exceptions
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java