public synchronized void start() throws IllegalStateException, StreamsException {
        if (setState(State.REBALANCING)) {
            log.debug("Starting Streams client");

            if (globalStreamThread != null) {
                globalStreamThread.start();
            }

            final int numThreads = processStreamThread(StreamThread::start);

            log.info("Started {} stream threads", numThreads);

            final Long cleanupDelay = applicationConfigs.getLong(StreamsConfig.STATE_CLEANUP_DELAY_MS_CONFIG);
            stateDirCleaner.scheduleAtFixedRate(() -> {
                // we do not use lock here since we only read on the value and act on it
                if (state == State.RUNNING) {
                    stateDirectory.cleanRemovedTasks(cleanupDelay);
                }
            }, cleanupDelay, cleanupDelay, TimeUnit.MILLISECONDS);

            final long recordingDelay = 0;
            final long recordingInterval = 1;
            if (rocksDBMetricsRecordingService != null) {
                rocksDBMetricsRecordingService.scheduleAtFixedRate(
                    streamsMetrics.rocksDBMetricsRecordingTrigger(),
                    recordingDelay,
                    recordingInterval,
                    TimeUnit.MINUTES
                );
            }
        } else {
            throw new IllegalStateException("The client is either already started or already stopped, cannot re-start");
        }
    }
Start the {@code KafkaStreams} instance by starting all its threads.
This function is expected to be called only once during the life cycle of the client.
<p>
Because threads are started in the background, this method does not block.
However, if you have global stores in your topology, this method blocks until all global stores are restored.
As a consequence, any fatal exception that happens during processing is by default only logged.
If you want to be notified about dying threads, you can
{@link #setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler) register an uncaught exception handler}
before starting the {@code KafkaStreams} instance.
<p>
Note, for brokers with version {@code 0.9.x} or lower, the broker version cannot be checked.
There will be no error and the client will hang and retry to verify the broker version until it
{@link StreamsConfig#REQUEST_TIMEOUT_MS_CONFIG times out}.
@throws IllegalStateException if process was already started
@throws StreamsException if the Kafka brokers have version 0.10.0.x or
                         if {@link StreamsConfig#PROCESSING_GUARANTEE_CONFIG exactly-once} is enabled for pre 0.11.0.x brokers
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java