private void setupTopology(final InternalTopologyBuilder builder,
                               final StreamsConfig streamsConfig) {
        internalTopologyBuilder = builder;
        internalTopologyBuilder.rewriteTopology(streamsConfig);

        processorTopology = internalTopologyBuilder.buildTopology();
        globalTopology = internalTopologyBuilder.buildGlobalStateTopology();

        for (final String topic : processorTopology.sourceTopics()) {
            final TopicPartition tp = new TopicPartition(topic, PARTITION_ID);
            partitionsByInputTopic.put(topic, tp);
            offsetsByTopicOrPatternPartition.put(tp, new AtomicLong());
        }

        stateDirectory = new StateDirectory(streamsConfig, mockWallClockTime, internalTopologyBuilder.hasPersistentStores(), false);
    }
Create a new test diver instance.

@param builder builder for the topology to be tested
@param config the configuration for the topology
@param initialWallClockTimeMs the initial value of internally mocked wall-clock time
streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java