Repository: apache/kafka
File: streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java (Java)

Code before:
                Task.TaskType.ACTIVE,
                StreamsConfig.EXACTLY_ONCE_V2.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.productionExceptionHandler(),
                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 -> { });
            for (final TopicPartition tp: task.inputPartitions()) {
                task.updateNextOffsets(tp, new OffsetAndMetadata(0, Optional.empty(), ""));
            }
        } else {
            task = null;
        }
    }


Review comment (suggestion):
I think `TopologyTestDriver` shouldn't set this value here as this RecordContext is being added in the downstream classes. 

Code after:
                TASK_ID,
                Task.TaskType.ACTIVE,
                StreamsConfig.EXACTLY_ONCE_V2.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.productionExceptionHandler(),
                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 -> { });
            for (final TopicPartition tp: task.inputPartitions()) {
                task.updateNextOffsets(tp, new OffsetAndMetadata(0, Optional.empty(), ""));
            }
        } else {
            task = null;
        }
    }