Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContext.java (Java)

Code before:
import static java.util.Objects.requireNonNull;
import static org.apache.kafka.common.utils.Utils.getNullableSizePrefixedArray;

public class ProcessorRecordContext implements RecordContext, RecordMetadata {

    private final long timestamp;
    private final long offset;
    private final String topic;
    private final int partition;
    private final Headers headers;
    private byte[] sourceRawKey;
    private byte[] sourceRawValue;

    public ProcessorRecordContext(final long timestamp,
                                  final long offset,
                                  final int partition,
                                  final String topic,
                                  final Headers headers) {
        this.timestamp = timestamp;
        this.offset = offset;
        this.topic = topic;
        this.partition = partition;
        this.headers = Objects.requireNonNull(headers);
        this.sourceRawKey = null;
        this.sourceRawValue = null;
    }

    public ProcessorRecordContext(final long timestamp,
                                  final long offset,
                                  final int partition,
                                  final String topic,
                                  final Headers headers,
                                  final byte[] sourceRawKey,
                                  final byte[] sourceRawValue) {
        this.timestamp = timestamp;
        this.offset = offset;
        this.topic = topic;
        this.partition = partition;
        this.headers = Objects.requireNonNull(headers);
        this.sourceRawKey = sourceRawKey;
        this.sourceRawValue = sourceRawValue;
    }

    @Override
    public long offset() {
        return offset;
    }

    @Override
    public long timestamp() {
        return timestamp;

Review comment (suggestion):
You also need to add these info to the `serialize()` and `deserialize()` so that the buffer values also get the source record. Here it gets a bit tricky, because you need to consider the case where a serialized record context does not contain the source record because it was written by a version of Streams that has not yet had the source record in the context.

Code after:
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Objects.requireNonNull;
import static org.apache.kafka.common.utils.Utils.getNullableSizePrefixedArray;

public class ProcessorRecordContext implements RecordContext, RecordMetadata {

    private final long timestamp;
    private final long offset;
    private final String topic;
    private final int partition;
    private final Headers headers;
    private byte[] sourceRawKey;
    private byte[] sourceRawValue;

    public ProcessorRecordContext(final long timestamp,
                                  final long offset,
                                  final int partition,
                                  final String topic,
                                  final Headers headers) {
        this.timestamp = timestamp;
        this.offset = offset;
        this.topic = topic;
        this.partition = partition;
        this.headers = Objects.requireNonNull(headers);
        this.sourceRawKey = null;
        this.sourceRawValue = null;
    }

    public ProcessorRecordContext(final long timestamp,
                                  final long offset,
                                  final int partition,
                                  final String topic,
                                  final Headers headers,
                                  final byte[] sourceRawKey,
                                  final byte[] sourceRawValue) {
        this.timestamp = timestamp;
        this.offset = offset;
        this.topic = topic;
        this.partition = partition;
        this.headers = Objects.requireNonNull(headers);
        this.sourceRawKey = sourceRawKey;
        this.sourceRawValue = sourceRawValue;
    }

    @Override
    public long offset() {
        return offset;
    }

    @Override
    public long timestamp() {