private void completeShutdown(final boolean cleanRun) {
        // set the state to pending shutdown first as it may be called due to error;
        // its state may already be PENDING_SHUTDOWN so it will return false but we
        // intentionally do not check the returned flag
        setState(State.PENDING_SHUTDOWN);

        log.info("Shutting down {}", cleanRun ? "clean" : "unclean");

        mainConsumerInstanceIdFuture.complete(null);

        try {
            taskManager.shutdown(cleanRun);
        } catch (final Throwable e) {
            log.error("Failed to close task manager due to the following error:", e);
        }
        try {
            topologyMetadata.unregisterThread(threadMetadata.threadName());
        } catch (final Throwable e) {
            log.error("Failed to unregister thread due to the following error:", e);
        }
        try {
            changelogReader.clear();
        } catch (final Throwable e) {
            log.error("Failed to close changelog reader due to the following error:", e);
        }
        try {
            if (leaveGroupRequested.get()) {
                mainConsumer.unsubscribe();
            }
        } catch (final Throwable e) {
            log.error("Failed to unsubscribe due to the following error: ", e);
        }
        try {
            mainConsumer.close();
        } catch (final Throwable e) {
            log.error("Failed to close consumer due to the following error:", e);
        }
        try {
            restoreConsumer.close();
        } catch (final Throwable e) {
            log.error("Failed to close restore consumer due to the following error:", e);
        }
        streamsMetrics.removeAllThreadLevelSensors(getName());
        streamsMetrics.removeAllThreadLevelMetrics(getName());

        setState(State.DEAD);

        log.info("Shutdown complete");
    }
Shutdown this stream thread.
<p>
Note that there is nothing to prevent this function from being called multiple times
(e.g., in testing), hence the state is set only the first time
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java