private void subscribeConsumer() {
        if (topologyMetadata.usesPatternSubscription()) {
            mainConsumer.subscribe(topologyMetadata.sourceTopicPattern(), rebalanceListener);
        } else {
            mainConsumer.subscribe(topologyMetadata.allFullSourceTopicNames(), rebalanceListener);
        }
    }
Sets the streams uncaught exception handler.

@param streamsUncaughtExceptionHandler the user handler wrapped in shell to execute the action
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java