[dvc] Rehydrate remote LCVP on restart and guard empty-buffer resolution

Address PR review feedback on the Global RT DIV remote-LCVP resume path:

- OffsetRecord#getLatestConsumedRemoteVtPosition now resolves a null/empty
  wire-format buffer to EARLIEST instead of deserializing an empty payload
  (which threw for pre-v24 persisted PartitionState records).
- PartitionTracker#setPartitionState rehydrates latestConsumedRemoteVtPosition
  from the durable OffsetRecord for the version topic on restart, so a
  follower's VT-DIV checkpoint no longer clobbers the persisted remote LCVP
  back to EARLIEST before an F->L resume.
- Fix TestAvroSchema#testStringMapInPartitionState to set the new v24
  upstreamLastConsumedVersionTopicPubSubPosition field.
- Add regression tests for both fixes.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: KaiSernLim

clients/da-vinci-client/src/main/java/com/linkedin/davinci/validation/PartitionTracker.java

@@ -179,6 +179,12 @@ public void setPartitionState(TopicType type, OffsetRecord offsetRecord, long ma
     long earliestAllowableTimestamp =
         computeEarliestAllowableTimestamp(maxAgeInMs, offsetRecord.calculateLatestMessageTimeInMs());
     setPartitionState(type, offsetRecord.getProducerPartitionStateMap(), earliestAllowableTimestamp);
+    if (TopicType.isVersionTopic(type)) {
+      // Rehydrate the durable remote LCVP so a follower's VT-DIV checkpoint writes back the persisted value instead of
+      // EARLIEST across a restart. Unlike the local LCVP (re-advanced live on the consume path), the remote LCVP is
+      // only advanced by a remote-consume leader, so without this it would collapse to EARLIEST before an F->L resume.
+      updateLatestConsumedRemoteVtPosition(offsetRecord.getLatestConsumedRemoteVtPosition());
+    }
   }
 
   public void setPartitionState(

clients/da-vinci-client/src/test/java/com/linkedin/davinci/validation/TestPartitionTracker.java

@@ -41,6 +41,7 @@
 import com.linkedin.venice.pubsub.adapter.kafka.common.ApacheKafkaOffsetPosition;
 import com.linkedin.venice.pubsub.api.DefaultPubSubMessage;
 import com.linkedin.venice.pubsub.api.PubSubPosition;
+import com.linkedin.venice.pubsub.api.PubSubSymbolicPosition;
 import com.linkedin.venice.pubsub.api.PubSubTopic;
 import com.linkedin.venice.pubsub.api.PubSubTopicPartition;
 import com.linkedin.venice.pubsub.mock.InMemoryPubSubPosition;
@@ -772,6 +773,29 @@ public void testUpdateOffsetRecordPersistsRemoteLcvp() {
     assertEquals(destTracker.getLatestConsumedRemoteVtPosition(), remoteLcvp, "remote LCVP should be copied on clone");
   }
 
+  /**
+   * On restart, {@link PartitionTracker#setPartitionState(TopicType, OffsetRecord, long)} must rehydrate the remote
+   * LCVP from the durable OffsetRecord. Otherwise a follower's VT-DIV checkpoint would write the in-memory EARLIEST
+   * back over the persisted value, collapsing the F->L remote-VT resume position to EARLIEST across any bounce.
+   */
+  @Test(timeOut = 10 * Time.MS_PER_SECOND)
+  public void testSetPartitionStateRehydratesRemoteLcvp() {
+    PubSubPosition remoteLcvp = ApacheKafkaOffsetPosition.of(100L);
+    OffsetRecord offsetRecord = TestUtils
+        .getOffsetRecord(ApacheKafkaOffsetPosition.of(0L), Optional.empty(), DEFAULT_PUBSUB_CONTEXT_FOR_UNIT_TESTING);
+    offsetRecord.setLatestConsumedRemoteVtPosition(remoteLcvp);
+
+    // Fresh tracker starts at EARLIEST, mirroring a restart before any remote-VT record is consumed.
+    PartitionTracker restartedTracker = createDestTracker();
+    assertEquals(restartedTracker.getLatestConsumedRemoteVtPosition(), PubSubSymbolicPosition.EARLIEST);
+
+    restartedTracker.setPartitionState(vt, offsetRecord, MAX_AGE_IN_MS);
+    assertEquals(
+        restartedTracker.getLatestConsumedRemoteVtPosition(),
+        remoteLcvp,
+        "Remote LCVP must be rehydrated from the OffsetRecord on restart so it is not clobbered back to EARLIEST");
+  }
+
   /**
    * Tests that cloneVtProducerStates correctly handles the maxAgeInMs threshold and data-relative anchor:
    * <ul>

internal/venice-common/src/main/java/com/linkedin/venice/offsets/OffsetRecord.java

@@ -394,8 +394,13 @@ public void setLatestConsumedRemoteVtPosition(PubSubPosition latestConsumedRemot
   }
 
   public PubSubPosition getLatestConsumedRemoteVtPosition() {
-    return pubSubPositionDeserializer
-        .toPosition(this.partitionState.getUpstreamLastConsumedVersionTopicPubSubPosition());
+    ByteBuffer wireFormat = this.partitionState.getUpstreamLastConsumedVersionTopicPubSubPosition();
+    // Records persisted before PartitionState v24 lack this field, so Avro schema resolution supplies the field
+    // default (empty bytes). Treat null/empty as EARLIEST rather than deserializing an empty payload, which throws.
+    if (wireFormat == null || !wireFormat.hasRemaining()) {
+      return PubSubSymbolicPosition.EARLIEST;
+    }
+    return pubSubPositionDeserializer.toPosition(wireFormat);
   }
 
   public long getActiveKeyCount() {

internal/venice-common/src/test/java/com/linkedin/venice/offsets/TestOffsetRecord.java

@@ -6,6 +6,7 @@
 import static org.testng.Assert.assertTrue;
 
 import com.linkedin.venice.kafka.protocol.GUID;
+import com.linkedin.venice.kafka.protocol.state.PartitionState;
 import com.linkedin.venice.kafka.protocol.state.ProducerPartitionState;
 import com.linkedin.venice.pubsub.adapter.kafka.common.ApacheKafkaOffsetPosition;
 import com.linkedin.venice.pubsub.api.PubSubPosition;
@@ -64,6 +65,19 @@ public void testToBytes() {
     Assert.assertEquals(offsetRecord2, offsetRecord1);
   }
 
+  @Test
+  public void testLatestConsumedRemoteVtPositionEmptyBufferDefaultsToEarliest() {
+    // Simulates a pre-v24 persisted record: Avro schema resolution supplies the field default (empty bytes) for the
+    // newly added field. The getter must resolve that to EARLIEST rather than deserializing an empty payload.
+    PartitionState partitionState = new PartitionState();
+    partitionState.upstreamLastConsumedVersionTopicPubSubPosition = ByteBuffer.allocate(0);
+    OffsetRecord offsetRecord = new OffsetRecord(
+        partitionState,
+        AvroProtocolDefinition.PARTITION_STATE.getSerializer(),
+        DEFAULT_PUBSUB_CONTEXT_FOR_UNIT_TESTING);
+    assertEquals(offsetRecord.getLatestConsumedRemoteVtPosition(), PubSubSymbolicPosition.EARLIEST);
+  }
+
   @Test
   public void testLatestConsumedRemoteVtPositionRoundTrip() {
     OffsetRecord fresh = new OffsetRecord(

internal/venice-common/src/test/java/com/linkedin/venice/schema/TestAvroSchema.java

@@ -75,6 +75,7 @@ public void testStringMapInPartitionState() throws IOException {
     ps.lastConsumedVersionTopicPubSubPosition = ByteBuffer.wrap("".getBytes());
     ps.upstreamRealTimeTopicPubSubPositionMap = new VeniceConcurrentHashMap<>();
     ps.upstreamVersionTopicPubSubPosition = ByteBuffer.wrap("".getBytes());
+    ps.upstreamLastConsumedVersionTopicPubSubPosition = ByteBuffer.wrap("".getBytes());
 
     AvroSerializer serializer = new AvroSerializer(ps.getSchema());
     byte[] serializedBytes = serializer.serialize(ps);