private void setupGlobalTask(final Time mockWallClockTime,
                                 final StreamsConfig streamsConfig,
                                 final StreamsMetricsImpl streamsMetrics,
                                 final ThreadCache cache) {
        if (globalTopology != null) {
            final MockConsumer<byte[], byte[]> globalConsumer = new MockConsumer<>(OffsetResetStrategy.NONE);
            for (final String topicName : globalTopology.sourceTopics()) {
                final TopicPartition partition = new TopicPartition(topicName, 0);
                globalPartitionsByInputTopic.put(topicName, partition);
                offsetsByTopicOrPatternPartition.put(partition, new AtomicLong());
                globalConsumer.updatePartitions(topicName, Collections.singletonList(
                    new PartitionInfo(topicName, 0, null, null, null)));
                globalConsumer.updateBeginningOffsets(Collections.singletonMap(partition, 0L));
                globalConsumer.updateEndOffsets(Collections.singletonMap(partition, 0L));
            }

            globalStateManager = new GlobalStateManagerImpl(
                logContext,
                mockWallClockTime,
                globalTopology,
                globalConsumer,
                stateDirectory,
                stateRestoreListener,
                streamsConfig
            );

            final GlobalProcessorContextImpl globalProcessorContext =
                new GlobalProcessorContextImpl(streamsConfig, globalStateManager, streamsMetrics, cache, mockWallClockTime);
            globalStateManager.setGlobalProcessorContext(globalProcessorContext);

            globalStateTask = new GlobalStateUpdateTask(
                logContext,
                globalTopology,
                globalProcessorContext,
                globalStateManager,
                new LogAndContinueExceptionHandler(),
                mockWallClockTime,
                streamsConfig.getLong(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG)
            );
            globalStateTask.initialize();
            globalProcessorContext.setRecordContext(null);
        } else {
            globalStateManager = null;
            globalStateTask = null;
        }
    }
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