Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java (Java)

Code before:
                globalThreadId,
                delegatingStateRestoreListener,
                exception -> handleStreamsUncaughtException(exception, t -> SHUTDOWN_CLIENT, false)
            );
            globalThreadState = globalStreamThread.state();
        }

        streamStateListener = new StreamStateListener(globalThreadState);

        final GlobalStateStoreProvider globalStateStoreProvider = new GlobalStateStoreProvider(this.topologyMetadata.globalStateStores());

        if (hasGlobalTopology) {
            globalStreamThread.setStateListener(streamStateListener);
        }

        queryableStoreProvider = new QueryableStoreProvider(globalStateStoreProvider);
        for (int i = 1; i <= numStreamThreads; i++) {
            createAndAddStreamThread(cacheSizePerThread, i);
        }

        stateDirCleaner = setupStateDirCleaner();
        rocksDBMetricsRecordingService = maybeCreateRocksDBMetricsRecordingService(clientId, applicationConfigs);
    }

    private void throwIfUnsupportedFeatureIsUsedWithStreamsRebalanceProtocol() {
        if (applicationConfigs.getString(StreamsConfig.GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.STREAMS.name)) {
            log.info("Streams rebalance protocol enabled");
            if (topologyMetadata.hasNamedTopologies()) {
                throw new UnsupportedOperationException("Named topologies are not supported with the STREAMS protocol.");
            }
            if (topologyMetadata.usesPatternSubscription()) {
                throw new UnsupportedOperationException("Pattern subscriptions are not supported with the STREAMS protocol.");
            }
            if (!(clientSupplier instanceof DefaultKafkaClientSupplier)) {
                throw new UnsupportedOperationException("A non-default kafka client supplier is not supported with the STREAMS protocol.");
            }
        }
    }

    private StreamThread createAndAddStreamThread(final long cacheSizePerThread, final int threadIdx) {
        final StreamThread streamThread = StreamThread.create(
            topologyMetadata,
            applicationConfigs,
            clientSupplier,
            adminClient,
            processId,
            clientId,
            streamsMetrics,
            time,
            streamsMetadataState,
            cacheSizePerThread,

Review comment (suggestion):
Consider invoking GroupProtocol.STREAMS.name() (with parentheses) instead of GroupProtocol.STREAMS.name to correctly compare the protocol string.
```suggestion
        if (applicationConfigs.getString(StreamsConfig.GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.STREAMS.name())) {
```

Code after:
                globalThreadId,
                delegatingStateRestoreListener,
                exception -> handleStreamsUncaughtException(exception, t -> SHUTDOWN_CLIENT, false)
            );
            globalThreadState = globalStreamThread.state();
        }

        streamStateListener = new StreamStateListener(globalThreadState);

        final GlobalStateStoreProvider globalStateStoreProvider = new GlobalStateStoreProvider(this.topologyMetadata.globalStateStores());

        if (hasGlobalTopology) {
            globalStreamThread.setStateListener(streamStateListener);
        }

        queryableStoreProvider = new QueryableStoreProvider(globalStateStoreProvider);
        for (int i = 1; i <= numStreamThreads; i++) {
            createAndAddStreamThread(cacheSizePerThread, i);
        }

        stateDirCleaner = setupStateDirCleaner();
        rocksDBMetricsRecordingService = maybeCreateRocksDBMetricsRecordingService(clientId, applicationConfigs);
    }

    private void throwIfUnsupportedFeatureIsUsedWithStreamsRebalanceProtocol() {
        if (applicationConfigs.isStreamsProtocolEnabled()) {
            log.info("Streams rebalance protocol enabled");
            if (topologyMetadata.hasNamedTopologies()) {
                throw new UnsupportedOperationException("Named topologies are not supported with the STREAMS protocol.");
            }
            if (topologyMetadata.usesPatternSubscription()) {
                throw new UnsupportedOperationException("Pattern subscriptions are not supported with the STREAMS protocol.");
            }
            if (!(clientSupplier instanceof DefaultKafkaClientSupplier)) {
                log.warn("A non-default kafka client supplier was supplied. Note that supplying a custom main consumer" +
                    " is not supported with the STREAMS protocol.");
            }
        }
    }

    private StreamThread createAndAddStreamThread(final long cacheSizePerThread, final int threadIdx) {
        final StreamThread streamThread = StreamThread.create(
            topologyMetadata,
            applicationConfigs,
            clientSupplier,
            adminClient,
            processId,
            clientId,
            streamsMetrics,
            time,
            streamsMetadataState,