@SuppressWarnings("deprecation")
    private void setupTask(final StreamsConfig streamsConfig,
                           final StreamsMetricsImpl streamsMetrics,
                           final ThreadCache cache,
                           final TaskConfig taskConfig) {
        if (!partitionsByInputTopic.isEmpty()) {
            consumer.assign(partitionsByInputTopic.values());
            final Map<TopicPartition, Long> startOffsets = new HashMap<>();
            for (final TopicPartition topicPartition : partitionsByInputTopic.values()) {
                startOffsets.put(topicPartition, 0L);
            }
            consumer.updateBeginningOffsets(startOffsets);

            final ProcessorStateManager stateManager = new ProcessorStateManager(
                TASK_ID,
                Task.TaskType.ACTIVE,
                StreamsConfig.EXACTLY_ONCE.equals(streamsConfig.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG)),
                logContext,
                stateDirectory,
                new MockChangelogRegister(),
                processorTopology.storeToChangelogTopic(),
                new HashSet<>(partitionsByInputTopic.values()),
                false);
            final RecordCollector recordCollector = new RecordCollectorImpl(
                logContext,
                TASK_ID,
                testDriverProducer,
                streamsConfig.defaultProductionExceptionHandler(),
                streamsMetrics,
                processorTopology
            );

            final InternalProcessorContext context = new ProcessorContextImpl(
                TASK_ID,
                streamsConfig,
                stateManager,
                streamsMetrics,
                cache
            );

            task = new StreamTask(
                TASK_ID,
                new HashSet<>(partitionsByInputTopic.values()),
                processorTopology,
                consumer,
                taskConfig,
                streamsMetrics,
                stateDirectory,
                cache,
                mockWallClockTime,
                stateManager,
                recordCollector,
                context,
                logContext,
                false
                );
            task.initializeIfNeeded();
            task.completeRestoration(noOpResetter -> { });
            task.processorContext().setRecordContext(null);

        } else {
            task = 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