[server] Attach vtp protocol-schema header on DoL stamp segment-start record

[sushantmane]

Summary

VeniceWriter.sendDoLStamp was passing EmptyPubSubMessageHeaders.SINGLETON directly to producerAdapter.sendMessage, bypassing the getHeaders() helper that attaches the VENICE_TRANSPORT_PROTOCOL_HEADER (vtp) on a segment-start record. Every DoL stamp therefore landed on the wire with empty headers — and when a STANDBY → LEADER transition writes a DoL stamp at offset 0 of a brand-new version topic, a forward-compat consumer whose jar predates the current KME protocol version has no on-record schema to bootstrap with. It throws:

com.linkedin.venice.exceptions.VeniceMessageException:
    Received Protocol Version 'N' which is not supported by KafkaValueSerializer.
    The only supported Protocol Versions are: {1,...,N-1}

Root cause

VeniceWriter.getHeaders(producerMetadata, addLeaderCompleteHeader, leaderCompleteState, extraHeaders) is the canonical helper that produces per-record PubSubMessageHeaders. Its contract:

• If producerMetadata.segmentNumber == 0 && producerMetadata.messageSequenceNumber == 0 (the first record of a new segment), it attaches the vtp header carrying the current KME protocol-schema JSON. The receiving consumer uses that bytes-blob to construct an Avro reader for the writer's protocol version, even if the receiver's jar doesn't know about it.
• Otherwise it returns extraHeaders unchanged.

sendHeartbeat already routes through getHeaders(...) and attaches vtp on the first heartbeat of every segment. sendDoLStamp did not — it constructed its KME via getDoLStampKME (which sets segmentNumber == 0 && messageSequenceNumber == 0) and then called producerAdapter.sendMessage with EmptyPubSubMessageHeaders.SINGLETON instead of routing through getHeaders. The DoL stamp's "I should be carrying vtp" precondition was always true, but vtp was never actually attached.

The bug was latent until KME v14 was activated (#2780). A consumer running an older jar (only supports v1..v13) that reads from offset 0 of a fresh version topic now hits a v14-encoded DoL stamp with no headers and no way to bootstrap — the deserialize fails permanently for that consumer until it's upgraded.

Fix

VeniceWriter.sendDoLStamp now routes its sendMessage call through:

getHeaders(
    kafkaMessageEnvelope.getProducerMetadata(),
    /* addLeaderCompleteHeader */ false,
    /* leaderCompleteState     */ null,
    EmptyPubSubMessageHeaders.SINGLETON)

This is the same pattern sendHeartbeat uses. Because DoL stamps always carry segmentNumber == 0 && messageSequenceNumber == 0 (set inside getDoLStampKME), the needVtpHeader branch fires and vtp is attached. No behavioral change for non-DoL paths.

Scope

This PR is intentionally minimal — only the writer fix and the regression test. A follow-up PR will add defense-in-depth on the consumer side by plumbing a KME SchemaReader through every callsite that opens its own PubSubMessageDeserializer, so consumers that land on a header-less record from some other future writer regression can still resolve an unknown protocol version. Splitting these makes the immediate fix easy to review and easy to backport.

Related

• [protocol] Add KafkaMessageEnvelope schema v14 to carry upstream record timestamp #2778 added KME schema v14 to the protocol.
• [compat][server] Activate KME v14 and stamp upstreamMessageTimestamp on leader produces #2780 activated KME v14 on the writer — exposed the latent missing-header bug.
• [server][dvc] Stop pinning ~16KB Avro schema header per queued PubSub message #2762 added vtp header stripping after successful deserialize (orthogonal — concerns memory retention, not schema bootstrap).
• [server] Gate vtp protocol-schema header behind VtpHeaderEmissionMode #2798 is a parallel guard on the heartbeat / SOS emission side.

Testing Done

• New unit test VeniceWriterUnitTest.testSendDoLStampCarriesVtpHeader: captures the headers passed to the producer adapter on a sendDoLStamp call and asserts (a) VENICE_TRANSPORT_PROTOCOL_HEADER is present, (b) the header bytes parse as the current KafkaMessageEnvelope protocol-schema JSON, and (c) the underlying producerMetadata.segmentNumber == 0 && messageSequenceNumber == 0 (the invariant that drives needVtpHeader=true inside getHeaders).
• Existing VeniceWriterUnitTest tests pass (testSendHeartbeat, testVeniceWriterCloseRetry, etc.).
• Compile-clean on internal:venice-common (main + test sources).

[Copilot]

Pull request overview

This PR fixes a forward-compatibility issue where VeniceWriter.sendDoLStamp produced a DoL stamp control message with empty headers, bypassing the getHeaders(...) helper that attaches the vtp (VENICE_TRANSPORT_PROTOCOL_HEADER) protocol-schema header needed for consumers to bootstrap unknown KafkaMessageEnvelope protocol versions.

Changes:

• Route sendDoLStamp through getHeaders(...) so segment-start DoL stamps carry the vtp protocol-schema header.
• Add a regression unit test asserting DoL stamps include vtp and that the DoL stamp’s ProducerMetadata triggers the segment-start header logic.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated no comments.

File	Description
internal/venice-common/src/main/java/com/linkedin/venice/writer/VeniceWriter.java	Ensures DoL stamp sends use getHeaders(...) so vtp is attached on segment-start records.
internal/venice-common/src/test/java/com/linkedin/venice/writer/VeniceWriterUnitTest.java	Adds regression coverage validating sendDoLStamp carries vtp and uses segment-start producer metadata.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[sushantmane]

Thanks for the review, @KaiSernLim!