public Optional<String> addStreamThread() {
        if (isRunningOrRebalancing()) {
            final StreamThread streamThread;
            synchronized (changeThreadCount) {
                final int threadIdx = getNextThreadIndex();
                final int numLiveThreads = getNumLiveStreamThreads();
                final long cacheSizePerThread = getCacheSizePerThread(numLiveThreads + 1);
                log.info("Adding StreamThread-{}, there will now be {} live threads and the new cache size per thread is {}",
                         threadIdx, numLiveThreads + 1, cacheSizePerThread);
                resizeThreadCache(cacheSizePerThread);
                // Creating thread should hold the lock in order to avoid duplicate thread index.
                // If the duplicate index happen, the metadata of thread may be duplicate too.
                streamThread = createAndAddStreamThread(cacheSizePerThread, threadIdx);
            }

            synchronized (stateLock) {
                if (isRunningOrRebalancing()) {
                    streamThread.start();
                    return Optional.of(streamThread.getName());
                } else {
                    log.warn("Terminating the new thread because the Kafka Streams client is in state {}", state);
                    streamThread.shutdown();
                    threads.remove(streamThread);
                    final long cacheSizePerThread = getCacheSizePerThread(getNumLiveStreamThreads());
                    log.info("Resizing thread cache due to terminating added thread, new cache size per thread is {}", cacheSizePerThread);
                    resizeThreadCache(cacheSizePerThread);
                    return Optional.empty();
                }
            }
        } else {
            log.warn("Cannot add a stream thread when Kafka Streams client is in state {}", state);
            return Optional.empty();
        }
    }
Adds and starts a stream thread in addition to the stream threads that are already running in this
Kafka Streams client.
<p>
Since the number of stream threads increases, the sizes of the caches in the new stream thread
and the existing stream threads are adapted so that the sum of the cache sizes over all stream
threads does not exceed the total cache size specified in configuration
{@link StreamsConfig#STATESTORE_CACHE_MAX_BYTES_CONFIG}.
<p>
Stream threads can only be added if this Kafka Streams client is in state RUNNING or REBALANCING.

@return name of the added stream thread or empty if a new stream thread could not be added
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java