private StreamThread createAndAddStreamThread(final long cacheSizePerThread, final int threadIdx) {
        final StreamThread streamThread = StreamThread.create(
            topologyMetadata,
            applicationConfigs,
            clientSupplier,
            adminClient,
            processId,
            clientId,
            streamsMetrics,
            time,
            streamsMetadataState,
            cacheSizePerThread,
            stateDirectory,
            delegatingStateRestoreListener,
            delegatingStandbyUpdateListener,
            threadIdx,
            KafkaStreams.this::closeToError,
            streamsUncaughtExceptionHandler
        );
        streamStateListener.registerStreamThread(streamThread);
        streamThread.setStateListener(streamStateListener);
        threads.add(streamThread);
        queryableStoreProvider.addStoreProviderForThread(streamThread.getName(), new StreamThreadStateStoreProvider(streamThread));
        return streamThread;
    }
Create a {@code KafkaStreams} instance.
<p>
Note: even if you never call {@link #start()} on a {@code KafkaStreams} instance,
you still must {@link #close()} it to avoid resource leaks.

@param topology       the topology specifying the computational logic
@param applicationConfigs         configs for Kafka Streams
@param time           {@code Time} implementation; cannot be null
@throws StreamsException if any fatal error occurs
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java