[server][common][admin-tool][vpj] Log non-null context for all PubSubPosition deserialization-failure paths

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/BlobTransferIngestionHelper.java

@@ -165,7 +165,7 @@ public boolean isReplicaLaggedAndNeedBlobTransfer(
       return isLagged;
     }
     if (offsetRecord.getOffsetLag() == 0
-        && PubSubSymbolicPosition.EARLIEST.equals(offsetRecord.getCheckpointedLocalVtPosition())) {
+        && PubSubSymbolicPosition.EARLIEST.equals(offsetRecord.getCheckpointedLocalVtPosition(replicaId))) {
       return true;
     }
     if (offsetRecord.getOffsetLag() < blobTransferDisabledOffsetLagThreshold) {

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/IngestionNotificationDispatcher.java

@@ -287,7 +287,7 @@ void reportDataRecoveryCompleted(PartitionConsumptionState pcs) {
         notifier -> notifier.dataRecoveryCompleted(
             topic,
             pcs.getPartition(),
-            pcs.getOffsetRecord().getCheckpointedLocalVtPosition(),
+            pcs.getOffsetRecord().getCheckpointedLocalVtPosition(pcs.getReplicaId()),
             ""));
   }
 

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/PartitionConsumptionState.java

@@ -453,8 +453,8 @@ public PartitionConsumptionState(
       divRtCheckpointPositions = new VeniceConcurrentHashMap<>(3);
       latestProcessedRtPositions = new VeniceConcurrentHashMap<>(3);
       if (offsetRecord.getLeaderTopic() != null && Version.isRealTimeTopic(offsetRecord.getLeaderTopic())) {
-        offsetRecord.cloneRtPositionCheckpoints(latestConsumedRtPositions);
-        offsetRecord.cloneRtPositionCheckpoints(latestProcessedRtPositions);
+        offsetRecord.cloneRtPositionCheckpoints(latestConsumedRtPositions, getReplicaId());
+        offsetRecord.cloneRtPositionCheckpoints(latestProcessedRtPositions, getReplicaId());
       }
       trackingIncrementalPushStatus = new VeniceConcurrentHashMap<>(3);
     } else {
@@ -469,8 +469,8 @@ public PartitionConsumptionState(
     this.localRegionName = localRegionName;
     // Restore in-memory latest consumed version topic position and leader info from the checkpoint version topic
     // position
-    this.latestProcessedVtPosition = offsetRecord.getCheckpointedLocalVtPosition();
-    this.latestProcessedRemoteVtPosition = offsetRecord.getCheckpointedRemoteVtPosition();
+    this.latestProcessedVtPosition = offsetRecord.getCheckpointedLocalVtPosition(getReplicaId());
+    this.latestProcessedRemoteVtPosition = offsetRecord.getCheckpointedRemoteVtPosition(getReplicaId());
     this.leaderHostId = offsetRecord.getLeaderHostId();
     this.leaderGUID = offsetRecord.getLeaderGUID();
     this.lastVTProduceCallFuture = CompletableFuture.completedFuture(null);
@@ -1351,7 +1351,7 @@ public PubSubPosition getLatestProcessedRtPosition(String pubSubBrokerAddress) {
        * will be updated, since those offset are not processed yet; so when leader try to get the upstream offsets for the very
        * first time, there are no records in {@link #latestProcessedRtPositions} yet.
        */
-      return getOffsetRecord().getCheckpointedRtPosition(pubSubBrokerAddress);
+      return getOffsetRecord().getCheckpointedRtPosition(pubSubBrokerAddress, getReplicaId());
     }
     return rtPosition;
   }

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java

@@ -1142,7 +1142,7 @@ public void kill() {
    */
   private boolean checkDatabaseIntegrity(OffsetRecord offsetRecord, PartitionConsumptionState pcs) {
     boolean returnStatus = true;
-    if (!PubSubSymbolicPosition.EARLIEST.equals(offsetRecord.getCheckpointedLocalVtPosition())) {
+    if (!PubSubSymbolicPosition.EARLIEST.equals(offsetRecord.getCheckpointedLocalVtPosition(pcs.getReplicaId()))) {
       StoreVersionState storeVersionState = storageEngine.getStoreVersionState();
       if (storeVersionState != null) {
         LOGGER.info("Found storeVersionState for replica: {}: checkDatabaseIntegrity will proceed", pcs.getReplicaId());
@@ -2630,7 +2630,8 @@ private void checkConsumptionStateWhenStart(
     // Once storage node restart, send the "START" status to controller to rebuild the task status.
     // If this storage node has never consumed data from this topic, instead of sending "START" here, we send it
     // once START_OF_PUSH message has been read.
-    if (!PubSubSymbolicPosition.EARLIEST.equals(offsetRecord.getCheckpointedLocalVtPosition())) {
+    if (!PubSubSymbolicPosition.EARLIEST
+        .equals(offsetRecord.getCheckpointedLocalVtPosition(newPartitionConsumptionState.getReplicaId()))) {
       StoreVersionState storeVersionState = storageEngine.getStoreVersionState();
       if (storeVersionState != null) {
         boolean sorted = storeVersionState.sorted;
@@ -2900,7 +2901,8 @@ private PartitionConsumptionState createAndInstallPartitionConsumptionState(PubS
         serverConfig.getRegionName());
     if (uniqueIngestedKeyCountHllEnabled) {
       int lgK = serverConfig.getUniqueIngestedKeyCountHllLog2K();
-      boolean isNewSubscription = PubSubSymbolicPosition.EARLIEST.equals(offsetRecord.getCheckpointedLocalVtPosition());
+      boolean isNewSubscription =
+          PubSubSymbolicPosition.EARLIEST.equals(offsetRecord.getCheckpointedLocalVtPosition(freshPcs.getReplicaId()));
       if (isNewSubscription) {
         freshPcs.initializeUniqueKeyCountHll(lgK);
       } else if (offsetRecord.getUniqueIngestedKeyCountHllSketch() != null) {
@@ -3704,7 +3706,7 @@ private void syncOffset(PartitionConsumptionState pcs) {
     if (!REDUNDANT_LOGGING_FILTER.isRedundantException(msg)) {
       final PubSubPosition position = (isGlobalRtDivEnabled())
           ? offsetRecord.getLatestConsumedVtPosition()
-          : offsetRecord.getCheckpointedLocalVtPosition();
+          : offsetRecord.getCheckpointedLocalVtPosition(pcs.getReplicaId());
       int percentage = -1;
       if (getServerConfig().isIngestionProgressLoggingEnabled()) {
         final PubSubTopicPartition topicPartition = pcs.getReplicaTopicPartition();
@@ -6103,7 +6105,8 @@ protected PubSubPosition extractUpstreamPosition(DefaultPubSubMessage consumerRe
       return PubSubUtil.deserializePositionWithOffsetFallback(
           leaderMetadataFooter.upstreamPubSubPosition,
           leaderMetadataFooter.upstreamOffset,
-          pubSubContext.getPubSubPositionDeserializer());
+          pubSubContext.getPubSubPositionDeserializer(),
+          getReplicaId(kafkaVersionTopic, consumerRecord.getPartition()));
     } else {
       // Directly use upstreamOffset without attempting position deserialization
       return PubSubUtil.fromKafkaOffset(leaderMetadataFooter.upstreamOffset);

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

@@ -32,6 +32,7 @@
 import com.linkedin.venice.utils.CollectionUtils;
 import com.linkedin.venice.utils.LatencyUtils;
 import com.linkedin.venice.utils.RedundantExceptionFilter;
+import com.linkedin.venice.utils.Utils;
 import com.linkedin.venice.utils.concurrent.VeniceConcurrentHashMap;
 import com.linkedin.venice.utils.lazy.Lazy;
 import java.nio.ByteBuffer;
@@ -1003,7 +1004,8 @@ private String generateMessage(
                 deserializePositionWithOffsetFallback(
                     consumerRecord.getValue().leaderMetadataFooter.upstreamPubSubPosition,
                     consumerRecord.getValue().leaderMetadataFooter.upstreamOffset,
-                    pubSubPositionDeserializer))
+                    pubSubPositionDeserializer,
+                    Utils.getReplicaId(consumerRecord.getTopicPartition())))
             .append("; upstream pub sub cluster ID: ")
             .append(consumerRecord.getValue().leaderMetadataFooter.upstreamKafkaClusterId)
             .append("; producer host name: ")

clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTaskTest.java

@@ -3592,7 +3592,7 @@ public void testShouldStartBlobTransferHybridStoreOffsetLagBelowThreshold() thro
     when(mockVeniceServerConfig.getBlobTransferDisabledTimeLagThresholdInMinutes()).thenReturn(0);
     // Offset lag below threshold — should bootstrap from Kafka
     when(mockOffset.getOffsetLag()).thenReturn(500L);
-    when(mockOffset.getCheckpointedLocalVtPosition()).thenReturn(new ApacheKafkaOffsetPosition(100L));
+    when(mockOffset.getCheckpointedLocalVtPosition(anyString())).thenReturn(new ApacheKafkaOffsetPosition(100L));
 
     assertFalse(leaderFollowerStoreIngestionTask.shouldStartBlobTransfer(0, "test_v1-0", mockConsumerAction));
   }
@@ -3609,7 +3609,7 @@ public void testShouldStartBlobTransferHybridStoreOffsetLagAboveThreshold() thro
     when(mockVeniceServerConfig.getBlobTransferDisabledTimeLagThresholdInMinutes()).thenReturn(0);
     // Offset lag above threshold — should use blob transfer
     when(mockOffset.getOffsetLag()).thenReturn(5000L);
-    when(mockOffset.getCheckpointedLocalVtPosition()).thenReturn(new ApacheKafkaOffsetPosition(100L));
+    when(mockOffset.getCheckpointedLocalVtPosition(anyString())).thenReturn(new ApacheKafkaOffsetPosition(100L));
 
     assertTrue(leaderFollowerStoreIngestionTask.shouldStartBlobTransfer(0, "test_v1-0", mockConsumerAction));
   }
@@ -3626,7 +3626,7 @@ public void testShouldStartBlobTransferHybridStoreZeroLagEarliestPosition() thro
     when(mockVeniceServerConfig.getBlobTransferDisabledTimeLagThresholdInMinutes()).thenReturn(0);
     // Zero offset lag but EARLIEST position — needs blob transfer
     when(mockOffset.getOffsetLag()).thenReturn(0L);
-    when(mockOffset.getCheckpointedLocalVtPosition()).thenReturn(PubSubSymbolicPosition.EARLIEST);
+    when(mockOffset.getCheckpointedLocalVtPosition(anyString())).thenReturn(PubSubSymbolicPosition.EARLIEST);
 
     assertTrue(leaderFollowerStoreIngestionTask.shouldStartBlobTransfer(0, "test_v1-0", mockConsumerAction));
   }

clients/venice-admin-tool/src/main/java/com/linkedin/venice/KafkaTopicDumper.java

@@ -503,7 +503,8 @@ private void logRecordMetadata(DefaultPubSubMessage record) {
               : PubSubUtil.deserializePositionWithOffsetFallback(
                   leaderMetadata.upstreamPubSubPosition,
                   leaderMetadata.upstreamOffset,
-                  pubSubPositionDeserializer),
+                  pubSubPositionDeserializer,
+                  Utils.getReplicaId(record.getTopicPartition())),
           leaderMetadata == null ? "-" : leaderMetadata.upstreamKafkaClusterId,
           chunkMetadata);
     } catch (Exception e) {
@@ -669,7 +670,8 @@ static String constructTopicSwitchLog(
             : PubSubUtil.deserializePositionWithOffsetFallback(
                 leaderMetadata.upstreamPubSubPosition,
                 leaderMetadata.upstreamOffset,
-                pubSubPositionDeserializer),
+                pubSubPositionDeserializer,
+                Utils.getReplicaId(record.getTopicPartition())),
         leaderMetadata == null ? "-" : leaderMetadata.upstreamKafkaClusterId);
   }
 

clients/venice-push-job/src/main/java/com/linkedin/venice/spark/consistency/LilyPadSnapshotBuilder.java

@@ -11,6 +11,7 @@
 import com.linkedin.venice.pubsub.api.PubSubPosition;
 import com.linkedin.venice.pubsub.api.PubSubTopicPartition;
 import com.linkedin.venice.utils.ByteUtils;
+import com.linkedin.venice.utils.Utils;
 import com.linkedin.venice.vpj.pubsub.input.PubSubSplitIterator;
 import java.io.IOException;
 import java.util.ArrayList;
@@ -80,7 +81,8 @@ public static LilyPadUtils.Snapshot<ComparablePubSubPosition> buildSnapshot(
         PubSubPosition rawPosition = PubSubUtil.deserializePositionWithOffsetFallback(
             leaderMetadata.upstreamPubSubPosition,
             leaderMetadata.upstreamOffset,
-            pubSubPositionDeserializer);
+            pubSubPositionDeserializer,
+            Utils.getReplicaId(topicPartition));
         ComparablePubSubPosition upstreamPosition = new ComparablePubSubPosition(rawPosition, consumer, topicPartition);
         ComparablePubSubPosition currentHw = runningHighWatermark.get(regionId);
         if (currentHw == null || upstreamPosition.compareTo(currentHw) > 0) {

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

@@ -134,9 +134,24 @@ public void clearPreviousStatusesEntry(CharSequence key) {
   }
 
   public PubSubPosition getCheckpointedLocalVtPosition() {
+    return getCheckpointedLocalVtPosition(null);
+  }
+
+  /**
+   * Same as {@link #getCheckpointedLocalVtPosition()}, but accepts a caller-supplied {@code replicaId} that is
+   * threaded down to {@link PubSubUtil#deserializePositionWithOffsetFallback} so a deserialization-failure warning
+   * can be attributed to the affected replica. {@code OffsetRecord} itself has no notion of which store-version and
+   * partition it belongs to (that context lives in the caller, e.g. {@code PartitionConsumptionState}), so callers
+   * that know their replica identity should pass it here instead of relying on the context-free overload.
+   *
+   * @param replicaId canonical replica identity (e.g. as produced by {@code Utils.getReplicaId}), or {@code null} if
+   *                  unavailable, in which case a static field-based label is logged instead
+   */
+  public PubSubPosition getCheckpointedLocalVtPosition(String replicaId) {
     return deserializePositionWithOffsetFallback(
         this.partitionState.lastProcessedVersionTopicPubSubPosition,
-        this.partitionState.offset);
+        this.partitionState.offset,
+        positionSourceOrDefault(replicaId, "OffsetRecord.lastProcessedVersionTopicPubSubPosition"));
   }
 
   public void checkpointLocalVtPosition(PubSubPosition vtPosition) {
@@ -149,9 +164,18 @@ public void checkpointLocalVtPosition(PubSubPosition vtPosition) {
   }
 
   public PubSubPosition getCheckpointedRemoteVtPosition() {
+    return getCheckpointedRemoteVtPosition(null);
+  }
+
+  /**
+   * Same as {@link #getCheckpointedRemoteVtPosition()}, but accepts a caller-supplied {@code replicaId}; see
+   * {@link #getCheckpointedLocalVtPosition(String)} for rationale.
+   */
+  public PubSubPosition getCheckpointedRemoteVtPosition(String replicaId) {
     return deserializePositionWithOffsetFallback(
         this.partitionState.upstreamVersionTopicPubSubPosition,
-        this.partitionState.upstreamVersionTopicOffset);
+        this.partitionState.upstreamVersionTopicOffset,
+        positionSourceOrDefault(replicaId, "OffsetRecord.upstreamVersionTopicPubSubPosition"));
   }
 
   public void checkpointRemoteVtPosition(PubSubPosition remoteVtPosition) {
@@ -305,13 +329,27 @@ public PubSubTopic getLeaderTopic(PubSubTopicRepository pubSubTopicRepository) {
    * call this API to get the latest upstream offset.
    */
   public PubSubPosition getCheckpointedRtPosition(String pubSubBrokerAddress) {
+    return getCheckpointedRtPosition(pubSubBrokerAddress, null);
+  }
+
+  /**
+   * Same as {@link #getCheckpointedRtPosition(String)}, but accepts a caller-supplied {@code replicaId}; see
+   * {@link #getCheckpointedLocalVtPosition(String)} for rationale. When no {@code replicaId} is available, the
+   * fallback label includes the broker address so the affected checkpoint field can still be pinpointed.
+   */
+  public PubSubPosition getCheckpointedRtPosition(String pubSubBrokerAddress, String replicaId) {
     Long offset = partitionState.upstreamOffsetMap.get(pubSubBrokerAddress);
     ByteBuffer wfBuffer = partitionState.upstreamRealTimeTopicPubSubPositionMap.get(pubSubBrokerAddress);
     if (offset == null) {
       // If the offset is not set, return EARLIEST symbolic position.
       return PubSubSymbolicPosition.EARLIEST;
     }
-    return deserializePositionWithOffsetFallback(wfBuffer, offset);
+    return deserializePositionWithOffsetFallback(
+        wfBuffer,
+        offset,
+        positionSourceOrDefault(
+            replicaId,
+            "OffsetRecord.upstreamRealTimeTopicPubSubPosition[" + pubSubBrokerAddress + "]"));
   }
 
   public void checkpointRtPosition(String pubSubBrokerAddress, PubSubPosition leaderPosition) {
@@ -335,14 +373,30 @@ public void checkpointRtPositions(@Nonnull Map<String, PubSubPosition> newRtPosi
    * Clone the checkpoint upstream positions map to another map provided as the input.
    */
   public void cloneRtPositionCheckpoints(@Nonnull Map<String, PubSubPosition> checkpointUpstreamPositionsReceiver) {
+    cloneRtPositionCheckpoints(checkpointUpstreamPositionsReceiver, null);
+  }
+
+  /**
+   * Same as {@link #cloneRtPositionCheckpoints(Map)}, but accepts a caller-supplied {@code replicaId}; see
+   * {@link #getCheckpointedLocalVtPosition(String)} for rationale.
+   */
+  public void cloneRtPositionCheckpoints(
+      @Nonnull Map<String, PubSubPosition> checkpointUpstreamPositionsReceiver,
+      String replicaId) {
     if (partitionState.upstreamOffsetMap != null && !partitionState.upstreamOffsetMap.isEmpty()) {
       Validate.notNull(checkpointUpstreamPositionsReceiver);
       checkpointUpstreamPositionsReceiver.clear();
       for (Map.Entry<String, Long> offsetEntry: partitionState.upstreamOffsetMap.entrySet()) {
         String pubSubBrokerAddress = offsetEntry.getKey();
         ByteBuffer wfBuffer = partitionState.upstreamRealTimeTopicPubSubPositionMap.get(pubSubBrokerAddress);
-        checkpointUpstreamPositionsReceiver
-            .put(pubSubBrokerAddress, deserializePositionWithOffsetFallback(wfBuffer, offsetEntry.getValue()));
+        checkpointUpstreamPositionsReceiver.put(
+            pubSubBrokerAddress,
+            deserializePositionWithOffsetFallback(
+                wfBuffer,
+                offsetEntry.getValue(),
+                positionSourceOrDefault(
+                    replicaId,
+                    "OffsetRecord.upstreamRealTimeTopicPubSubPosition[" + pubSubBrokerAddress + "]")));
       }
     }
   }
@@ -551,10 +605,35 @@ public byte[] toBytes() {
    */
   @VisibleForTesting
   PubSubPosition deserializePositionWithOffsetFallback(ByteBuffer wireFormatBytes, long offset) {
+    return deserializePositionWithOffsetFallback(wireFormatBytes, offset, "OffsetRecord");
+  }
+
+  /**
+   * Same as {@link #deserializePositionWithOffsetFallback(ByteBuffer, long)}, but forwards an explicit
+   * {@code positionSource} label down to {@link PubSubUtil#deserializePositionWithOffsetFallback} so that a
+   * deserialization-failure warning can be attributed to the checkpoint field (and replica, when known) it came
+   * from, instead of logging a bare "N/A".
+   */
+  private PubSubPosition deserializePositionWithOffsetFallback(
+      ByteBuffer wireFormatBytes,
+      long offset,
+      String positionSource) {
     if (pubSubContext != null && !pubSubContext.isUseCheckpointedPubSubPositionWithFallbackEnabled()) {
       // When feature flag is disabled, use offset-only approach without attempting wire format deserialization
       return PubSubUtil.fromKafkaOffset(offset);
     }
-    return PubSubUtil.deserializePositionWithOffsetFallback(wireFormatBytes, offset, pubSubPositionDeserializer);
+    return PubSubUtil
+        .deserializePositionWithOffsetFallback(wireFormatBytes, offset, pubSubPositionDeserializer, positionSource);
+  }
+
+  /**
+   * Returns {@code replicaId} when non-null, otherwise {@code fallbackLabel}. Used so that every internal call to
+   * {@link PubSubUtil#deserializePositionWithOffsetFallback} always supplies a non-null, meaningful diagnostic
+   * string: the real replica identity when a caller provides one, or a stable field-based label identifying which
+   * checkpoint accessor was invoked when it doesn't (e.g. {@code toString()}, test/tooling callers that predate
+   * replica-aware plumbing).
+   */
+  private static String positionSourceOrDefault(String replicaId, String fallbackLabel) {
+    return replicaId != null ? replicaId : fallbackLabel;
   }
 }