[da-vinci] Lazily discover server blob fallback peers

[shresthhh]

Problem Statement

ServerAndDaVinciBlobFinder eagerly fetched Venice server metadata for every partition, even when a Da Vinci peer was available and could serve the blob. The server metadata response contains routing and schema information for the entire store, so eager requests during a partitioned cold start can add unnecessary storage-node load and affect Fast Client read latency.

Solution

Introduce lazy, tiered peer discovery:

• Discover and attempt Da Vinci peers first.
• Discover Venice server peers only when no Da Vinci peers are available or every primary peer transfer fails.
• Stop before fallback discovery when the partition transfer is cancelled.
• Preserve existing randomization within each peer tier.

The behavior remains gated by the existing davinci.blob.transfer.server.fallback.enabled config, which defaults to false.

Always resolving the partition transfer future

Fallback discovery is dispatched onto replicaBlobFetchExecutor rather than running inline in get(), which makes failure paths reachable that the eager implementation could not hit. A partition transfer future that is never completed stalls the replica forever, because StoreIngestionTask waits on it to decide whether to fall back to Kafka bootstrapping. Each path is now resolved explicitly:

• Guard the fallback dispatch and the discovery body, so a rejected execution (close() shutting the executor down) or an unexpected runtime failure still completes the future.
• Observe the peer chain with exceptionally(). A chain that completes exceptionally never reaches thenRun, so the transfer previously hung. This also covers the pre-existing single-tier path, where the same window existed but was only reachable during shutdown.
• Report a requested cancellation from these handlers instead of claiming every peer was tried, matching what the normal peer-exhaustion path reports.
• Log the fallback discovery error message rather than dropping it.

Blob transfer time and throughput are still measured per discovery pass, so recordBlobTransferTimeInSec and recordBlobTransferFileReceiveThroughput keep their existing meaning on the default (fallback-disabled) path.

Code changes

• Added new code behind a config. Existing config: davinci.blob.transfer.server.fallback.enabled (default: false).
• Introduced new log lines.
  • Confirmed if logs need to be rate limited to avoid excessive logging. Primary discovery retains the existing per-partition logging pattern; server discovery logs only when fallback is actually attempted. The failure log lines are terminal for a partition-level transfer, so they cannot repeat within a pass.

Concurrency-Specific Checks

Both reviewer and PR author to verify

• Code has no race conditions or thread safety issues. Cancellation is re-checked at each hand-off, including after a rejected dispatch, so a cancellation racing a failure is reported as a cancellation.
• Proper synchronization mechanisms (e.g., synchronized, RWLock) are used where needed. State is confined to the per-partition CompletableFuture, whose completion is idempotent, so no additional locking is introduced.
• No blocking calls inside critical sections that could lead to deadlocks or performance degradation. Fallback discovery blocks on the metadata fetch, so it runs on replicaBlobFetchExecutor instead of a Netty event-loop thread, consistent with the existing thenComposeAsync rationale.
• Verified thread-safe collections are used (e.g., ConcurrentHashMap, CopyOnWriteArrayList).
• Validated proper exception handling in multi-threaded code to avoid silent thread termination. Work dispatched to the executor is wrapped so an escaping exception cannot be swallowed by an unobserved future; Error is deliberately left to propagate.

How was this PR tested?

• Local code review completed.
• New unit tests added.
• New integration tests added.
• Modified or extended existing tests.
• Verified backward compatibility (if applicable).

New TestNettyP2PBlobTransferManager cases:

• testDoesNotDiscoverFallbackAfterPrimaryTransferSucceeds - server discovery is skipped when a Da Vinci peer serves the blob.
• testDiscoversFallbackWhenNoPrimaryPeersExist and testFallsBackFromDaVinciPeerToServerAfterTransferFailure - both fallback triggers, asserting discovery and transfer ordering.
• testCancellationDoesNotDiscoverFallback - a cancelled transfer stops before fallback discovery.
• testFallbackDiscoveryErrorAfterPrimaryPeersFail - fallback discovery returning an error surfaces VenicePeersAllFailedException.
• testFallbackDispatchRejectionFailsTransfer and testPeerChainRejectionMidFlightFailsTransfer - both executor-rejection windows resolve the future instead of hanging.
• testCancellationWinsOverChainRejection - a rejection racing a cancellation reports VeniceBlobTransferCancelledException.

Each new failure-path test was confirmed to fail against the unfixed code before the fix was applied.

Commands:

• ./gradlew :internal:venice-common:test --tests 'com.linkedin.venice.blobtransfer.ServerAndDaVinciBlobFinderTest'
• ./gradlew :clients:da-vinci-client:test --tests 'com.linkedin.davinci.blobtransfer.TestNettyP2PBlobTransferManager'
• ./gradlew spotlessCheck

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.

[Copilot]

Pull request overview

This PR reduces storage-node load during Da Vinci cold starts by changing blob peer discovery from an eager “Da Vinci + server” fetch to a lazy, tiered approach: try Da Vinci peers first, and only discover Venice server peers when needed (and skip fallback if the transfer is cancelled).

Changes:

• Introduces a fallback-capable BlobFinder contract (supportsFallback + discoverFallbackBlobPeers) and updates ServerAndDaVinciBlobFinder to only discover server peers on-demand.
• Updates NettyP2PBlobTransferManager to process primary peers first, then trigger fallback discovery only after primary peers are exhausted or unavailable, with cancellation-aware short-circuiting.
• Expands unit tests to cover fallback-after-failure, no-fallback-after-success, fallback-when-no-primary-peers, and cancellation behavior.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
internal/venice-common/src/test/java/com/linkedin/venice/blobtransfer/ServerAndDaVinciBlobFinderTest.java	Updates tests for the new “primary-only + explicit fallback” discovery API.
internal/venice-common/src/main/java/com/linkedin/venice/blobtransfer/ServerAndDaVinciBlobFinder.java	Makes server peer discovery lazy via discoverFallbackBlobPeers.
internal/venice-common/src/main/java/com/linkedin/venice/blobtransfer/BlobFinder.java	Adds fallback discovery methods to the BlobFinder contract.
clients/da-vinci-client/src/test/java/com/linkedin/davinci/blobtransfer/TestNettyP2PBlobTransferManager.java	Adds test coverage for tiered discovery and cancellation interactions.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/blobtransfer/NettyP2PBlobTransferManager.java	Implements tiered primary/fallback discovery and cancellation-aware fallback suppression.

---

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

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

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

Comments suppressed due to low confidence (1)

clients/da-vinci-client/src/main/java/com/linkedin/davinci/blobtransfer/NettyP2PBlobTransferManager.java:136

• processFallbackPeers(...) can run inline on the caller thread when primary discovery is unavailable (lines 129-135). That method immediately calls peerFinder.discoverFallbackBlobPeersIfEnabled(...), which may perform synchronous network/metadata fetch work (e.g., MetadataBasedServerBlobFinder.discoverBlobPeers blocks on metadata fetch). This reintroduces the same risk this class documents for Netty event-loop threads (see the thenComposeAsync rationale) and also bypasses the rejection-guard logic used when fallback is triggered after primary peers are exhausted.

Dispatch fallback discovery onto replicaBlobFetchExecutor in this branch as well, and handle RejectedExecutionException/other RuntimeException the same way as the peers-exhausted path so the partition future is always resolved.

    if (isDiscoveryUnavailable(response)) {
      if (peerFinder.supportsFallback()) {
        processFallbackPeers(storeName, version, partition, tableFormat, perPartitionTransferFuture, false);
      } else {
        completeWithNoPeersFound(replicaId, perPartitionTransferFuture);