private void closeToError() {
        if (!setState(State.PENDING_ERROR)) {
            log.info("Skipping shutdown since we are already in " + state());
        } else {
            final Thread shutdownThread = shutdownHelper(true, -1, false);

            shutdownThread.setDaemon(true);
            shutdownThread.start();
        }
    }
Shutdown this {@code KafkaStreams} instance by signaling all the threads to stop, and then wait for them to join.
This will block until all threads have stopped.
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java