private void replaceStreamThread(final Throwable throwable) {
        if (globalStreamThread != null && Thread.currentThread().getName().equals(globalStreamThread.getName())) {
            log.warn("The global thread cannot be replaced. Reverting to shutting down the client.");
            log.error("Encountered the following exception during processing " +
                    " The streams client is going to shut down now. ", throwable);
            closeToError();
        }
        final StreamThread deadThread = (StreamThread) Thread.currentThread();
        deadThread.shutdown();
        addStreamThread();
        if (throwable instanceof RuntimeException) {
            throw (RuntimeException) throwable;
        } else if (throwable instanceof Error) {
            throw (Error) throwable;
        } else {
            throw new RuntimeException("Unexpected checked exception caught in the uncaught exception handler", throwable);
        }
    }
Set the handler invoked when an internal {@link StreamsConfig#NUM_STREAM_THREADS_CONFIG stream thread}
throws an unexpected exception.
These might be exceptions indicating rare bugs in Kafka Streams, or they
might be exceptions thrown by your code, for example a NullPointerException thrown from your processor logic.
The handler will execute on the thread that produced the exception.
In order to get the thread that threw the exception, use {@code Thread.currentThread()}.
<p>
Note, this handler must be threadsafe, since it will be shared among all threads, and invoked from any
thread that encounters such an exception.

@param userStreamsUncaughtExceptionHandler the uncaught exception handler of type {@link StreamsUncaughtExceptionHandler} for all internal threads
@throws IllegalStateException if this {@code KafkaStreams} instance has already been started.
@throws NullPointerException if userStreamsUncaughtExceptionHandler is null.
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java