[server][da-vinci][test] Optimize Heartbeat Messages by Excluding VTP Headers #2295

lluwm · 2025-11-17T21:25:18Z

Problem Statement

Heartbeat messages in Venice were including Venice Transport Protocol (VTP) headers unnecessarily, adding overhead to frequent heartbeat communications. Since heartbeats are primarily for liveness detection rather than data transport, this overhead was wasteful.

Solution

Implemented an optimization to conditionally exclude VTP headers from heartbeat messages when Venice's robust schema evolution infrastructure is available to handle compatibility.

When dependent features are enabled, heartbeat messages skip VTP headers because:

KME Schema Registration: Servers register schemas during startup via ControllerClientBackedSystemSchemaInitializer
Schema Reader: RouterBackedSchemaReader automatically fetches unknown schemas from system store
Graceful Handling: InternalAvroSpecificSerializer handles unknown protocol versions via schema reader

Testing

Updated existing tests to maintain compatibility
Add new integration test to validate that optimization works correctly with Venice's KME schema evolution infrastructure (TestServerKMERegistrationFromMessageHeader.java)

Code changes

Added new code behind a config. If so list the config names and their default values in the PR description.
Introduced new log lines.
- Confirmed if logs need to be rate limited to avoid excessive logging.

Concurrency-Specific Checks

Both reviewer and PR author to verify

Code has no race conditions or thread safety issues.
Proper synchronization mechanisms (e.g., synchronized, RWLock) are used where needed.
No blocking calls inside critical sections that could lead to deadlocks or performance degradation.
Verified thread-safe collections are used (e.g., ConcurrentHashMap, CopyOnWriteArrayList).
Validated proper exception handling in multi-threaded code to avoid silent thread termination.

How was this PR tested?

New unit tests added.
New integration tests added.
Modified or extended existing tests.
Verified backward compatibility (if applicable).

Does this PR introduce any user-facing or breaking changes?

No. You can skip the rest of this section.
Yes. Clearly explain the behavior change and its impact.

… Headers Problem Statement Heartbeat messages in Venice were including Venice Transport Protocol (VTP) headers unnecessarily, adding overhead to frequent heartbeat communications. Since heartbeats are primarily for liveness detection rather than data transport, this overhead was wasteful. Solution Implemented an optimization to conditionally exclude VTP headers from heartbeat messages when Venice's robust schema evolution infrastructure is available to handle compatibility. When dependent features are enabled, heartbeat messages skip VTP headers because: - KME Schema Registration: Servers register schemas during startup via ControllerClientBackedSystemSchemaInitializer - Schema Reader: RouterBackedSchemaReader automatically fetches unknown schemas from system store - Graceful Handling: InternalAvroSpecificSerializer handles unknown protocol versions via schema reader Testing - Updated existing tests to maintain compatibility - Add new integration test to validate that optimization works correctly with Venice's KME schema evolution infrastructure (TestServerKMERegistrationFromMessageHeader.java)

Copilot

Pull Request Overview

This PR optimizes heartbeat messages in Venice by conditionally excluding Venice Transport Protocol (VTP) headers when the system's schema evolution infrastructure can handle compatibility. The optimization reduces overhead for frequent heartbeat communications while maintaining schema evolution capabilities through KME schema registration and RouterBackedSchemaReader.

Key changes:

Added logic to conditionally skip VTP headers in heartbeat messages based on dependent feature flags
Updated VeniceWriter to accept a new parameter controlling VTP header inclusion
Extended test coverage with new integration tests and updated existing mocks

Reviewed Changes

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

Show a summary per file

File	Description
VeniceWriter.java	Refactored header logic to conditionally add VTP headers based on message type and feature flags
StoreIngestionTask.java	Added storage for system schema initialization flag
LeaderFollowerStoreIngestionTask.java	Integrated dependent feature check before sending heartbeats
KafkaStoreIngestionService.java	Added method to check KME schema reader presence and improved logging
HeartbeatMonitoringService.java	Added getter for KafkaStoreIngestionService
VeniceWriterUnitTest.java	Updated test to cover VTP header skip logic with additional test parameters
VeniceWriterHeartbeatHeaderTest.java	New test file validating VTP header behavior for heartbeats vs data messages
TestServerKMERegistrationFromMessageHeader.java	Added integration test for heartbeat optimization and refactored common setup
MaterializedViewWriterTest.java	Updated mocks to include new heartbeat parameter
StoreIngestionTaskTest.java	Updated mocks to include new heartbeat parameter
LeaderFollowerStoreIngestionTaskTest.java	Added comprehensive test for dependent feature flag combinations

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

...nci-client/src/main/java/com/linkedin/davinci/kafka/consumer/KafkaStoreIngestionService.java

Copilot · 2025-11-17T21:35:11Z

internal/venice-common/src/main/java/com/linkedin/venice/writer/VeniceWriter.java

+     *
+     * Server 1 deploys -> Registers KME v2 in system store during startup. (requires isSystemSchemaInitializationAtStartTimeEnabled = true)
+     * Server 1 sends heartbeat -> Serialized with KME v2 (no VTP header)
+     * Server 2 receives heartbeat -> Detects unknown protocol version, schema reader fetches KME v2. (requires KME scheme reader to be present)


Corrected spelling of 'scheme' to 'schema' in comment.

Suggested change

* Server 2 receives heartbeat -> Detects unknown protocol version, schema reader fetches KME v2. (requires KME scheme reader to be present)

* Server 2 receives heartbeat -> Detects unknown protocol version, schema reader fetches KME v2. (requires KME schema reader to be present)

Copilot

Pull Request Overview

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

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

Copilot · 2025-11-18T18:43:14Z

...tegrationTest/java/com/linkedin/venice/helix/TestServerKMERegistrationFromMessageHeader.java

+    for (int i = 0; i < 10; i++) {
+      veniceWriter.sendHeartbeat(
+          topicPartition,
+          null, // No callback needed for this test


[nitpick] Passing null for the callback is acceptable here, but it would be more explicit to use a mock callback or a no-op callback to better demonstrate intent and avoid potential NPE issues if the implementation changes.

Suggested change

null, // No callback needed for this test

(metadata, exception) -> {}, // No-op callback for clarity and safety

Copilot · 2025-11-18T18:43:15Z

...ient/src/main/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTask.java

        leaderCompleteState,
        originTimeStampMs);
  }



[nitpick] Changing the visibility of 'sendIngestionHeartbeat' from private to package-private for testing purposes is acceptable, but consider adding a comment explaining that this is intentionally package-private for test access to make the design decision explicit.

Suggested change

/**

* Package-private for test access. This method's visibility is intentionally not private

* to allow unit tests in the same package to invoke it.

*/

Copilot · 2025-11-18T18:43:15Z

...nci-client/src/main/java/com/linkedin/davinci/kafka/consumer/KafkaStoreIngestionService.java

+    kafkaMessageEnvelopeSchemaReader.ifPresent(reader -> {
+      LOGGER.info(
+          "Initialized KME schema reader. Type: {}, Latest value schema ID: {}",
+          reader.getClass().getSimpleName(),
+          reader.getLatestValueSchemaId());
+      kafkaValueSerializer.setSchemaReader(reader);


[nitpick] The log message at INFO level is logged every time the service initializes. Consider whether this should be rate-limited or moved to DEBUG level to avoid log pollution in environments with frequent restarts, or ensure it's only logged once during service lifetime.

Copilot AI review requested due to automatic review settings November 17, 2025 21:25

Copilot started reviewing on behalf of lluwm November 17, 2025 21:25 View session

Copilot finished reviewing on behalf of lluwm November 17, 2025 21:26

Copilot AI reviewed Nov 17, 2025

View reviewed changes

lluwm added 2 commits November 17, 2025 15:19

fix typos

972eceb

try fix ingestion isolation

4f24d50

Copilot AI review requested due to automatic review settings November 18, 2025 18:32

Copilot started reviewing on behalf of lluwm November 18, 2025 18:33 View session

Copilot finished reviewing on behalf of lluwm November 18, 2025 18:34

Copilot AI reviewed Nov 18, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[server][da-vinci][test] Optimize Heartbeat Messages by Excluding VTP Headers #2295

[server][da-vinci][test] Optimize Heartbeat Messages by Excluding VTP Headers #2295

Uh oh!

lluwm commented Nov 17, 2025 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Copilot AI Nov 17, 2025

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Nov 18, 2025

Uh oh!

Copilot AI Nov 18, 2025

Uh oh!

Copilot AI Nov 18, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

	* Server 2 receives heartbeat -> Detects unknown protocol version, schema reader fetches KME v2. (requires KME scheme reader to be present)
	* Server 2 receives heartbeat -> Detects unknown protocol version, schema reader fetches KME v2. (requires KME schema reader to be present)

	null, // No callback needed for this test
	(metadata, exception) -> {}, // No-op callback for clarity and safety

+  /**
+   * Package-private for test access. This method's visibility is intentionally not private
+   * to allow unit tests in the same package to invoke it.
+   */

[server][da-vinci][test] Optimize Heartbeat Messages by Excluding VTP Headers #2295

Are you sure you want to change the base?

[server][da-vinci][test] Optimize Heartbeat Messages by Excluding VTP Headers #2295

Uh oh!

Conversation

lluwm commented Nov 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Problem Statement

Solution

Code changes

Concurrency-Specific Checks

How was this PR tested?

Does this PR introduce any user-facing or breaking changes?

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Uh oh!

Copilot AI Nov 17, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Uh oh!

Copilot AI Nov 18, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 18, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 18, 2025

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

lluwm commented Nov 17, 2025 •

edited

Loading