[server] Gate vtp protocol-schema header behind VtpHeaderEmissionMode

[sushantmane]

Summary

VeniceWriter attaches the vtp protocol-schema header (the Avro JSON for
KafkaMessageEnvelope, ~16 KB) on the first message of every segment
(segmentNumber == 0 && messageSequenceNumber == 0). Heartbeat control messages
are constructed as START_OF_SEGMENT with both numbers at zero, so under the
pre-existing rule every heartbeat picks up the ~16 KB header even though
heartbeat consumers never use it for schema bootstrap.

On busy ingestion paths with many partitions and frequent heartbeats this
dominates the consumer-side per-record memory footprint: a small data record
can share an in-flight queue with thousands of ~16 KB heartbeat envelopes, and
the queue size becomes a function of vtp retention rather than payload size.

This PR introduces a writer-scoped enum, VtpHeaderEmissionMode, surfaced via
the new config key venice.writer.vtp.header.emission.mode:

Mode	Behavior
SOS_AND_HB (default)	Emit on every segment-start, including heartbeats. Preserves pre-existing behavior — nothing changes on upgrade.
SOS_ONLY	Emit on regular data segment-start records only; skip heartbeat SOS. Consumers must resolve the KafkaMessageEnvelope schema by some other means (earlier data SOS on the same segment, or an out-of-band schema cache).
NONE	Never emit. Use only when all consumers can resolve the schema without the per-segment hint.

Changes

• New enum com.linkedin.venice.writer.VtpHeaderEmissionMode documenting the
  three modes.
• New config key VENICE_WRITER_VTP_HEADER_EMISSION_MODE
  (venice.writer.vtp.header.emission.mode) in ConfigKeys.
• VeniceWriter parses the property in its constructor; unrecognized values
  fall back to SOS_AND_HB with a warning log, matching the rest of the
  writer's tolerant config parsing.
• VeniceWriter#getHeaders gains a boolean isHeartbeat parameter so the
  call site can distinguish heartbeat SOS from data SOS. The three existing
  invocations (one in the regular send path, two in sendHeartbeat) are
  updated in place.
• The pre-existing gate on protocolSchemaHeader != null is unchanged.

Testing Done

• testHeartbeatVtpEmissionMode (TestNG DataProvider over all three modes)
  verifies the vtp header is present on the heartbeat under SOS_AND_HB and
  absent under SOS_ONLY / NONE.
• testDataSosWithVtpEmissionModeNone verifies that NONE also drops the
  vtp header on regular data segment-start records — not just heartbeats —
  so consumers can rely on the documented semantics.
• Full VeniceWriterUnitTest class passes (regression check for the new
  isHeartbeat parameter threading).

> Task :internal:venice-common:test
com.linkedin.venice.writer.VeniceWriterUnitTest > testDataSosWithVtpEmissionModeNone PASSED (523 ms)
com.linkedin.venice.writer.VeniceWriterUnitTest > testHeartbeatVtpEmissionMode[0](SOS_AND_HB, true) PASSED (122 ms)
com.linkedin.venice.writer.VeniceWriterUnitTest > testHeartbeatVtpEmissionMode[1](SOS_ONLY, false) PASSED (3 ms)
com.linkedin.venice.writer.VeniceWriterUnitTest > testHeartbeatVtpEmissionMode[2](NONE, false) PASSED (3 ms)
BUILD SUCCESSFUL in 9s

Backward Compatibility

Default is SOS_AND_HB, which preserves the pre-existing rule
(isFirstMessageOfFirstSegment). Existing deployments see no behavior change
unless they explicitly opt into SOS_ONLY or NONE.

[Copilot]

Pull request overview

This PR adds a writer-scoped configuration to control whether VeniceWriter emits the vtp (transport protocol schema) header on segment-start messages, allowing operators to avoid attaching the ~16KB KafkaMessageEnvelope schema blob to heartbeat SOS records.

Changes:

• Introduces VtpHeaderEmissionMode (SOS_AND_HB default, SOS_ONLY, NONE) and wires it into VeniceWriter header construction (including a new isHeartbeat parameter).
• Adds config key venice.writer.vtp.header.emission.mode to control the emission mode (tolerant parsing with warn-and-default behavior).
• Adds unit tests verifying heartbeat behavior across modes and that NONE suppresses vtp on data SOS as well.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 3 comments.

File	Description
internal/venice-common/src/test/java/com/linkedin/venice/writer/VeniceWriterUnitTest.java	Adds tests validating vtp header emission on heartbeats and data SOS under different modes.
internal/venice-common/src/main/java/com/linkedin/venice/writer/VtpHeaderEmissionMode.java	New enum defining emission modes and documenting intended semantics.
internal/venice-common/src/main/java/com/linkedin/venice/writer/VeniceWriter.java	Parses the new config and gates vtp header attachment based on mode + heartbeat vs data SOS.
internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java	Adds the new config key and Javadoc describing how to use it.

---

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