[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/StoreIngestionTask.java

@@ -6103,7 +6103,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);

internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubUtil.java

@@ -367,6 +367,28 @@ public static PubSubPosition deserializePositionWithOffsetFallback(
       ByteBuffer wireFormatBytes,
       long offset,
       PubSubPositionDeserializer pubSubPositionDeserializer) {
+    return deserializePositionWithOffsetFallback(wireFormatBytes, offset, pubSubPositionDeserializer, null);
+  }
+
+  /**
+   * Same as {@link #deserializePositionWithOffsetFallback(ByteBuffer, long, PubSubPositionDeserializer)}, but also
+   * accepts a {@code replicaId} (canonical {@code <store>_v<version>-<partition>} identity, e.g. as produced by
+   * {@code Utils.getReplicaId}) that is included in the warning logged on deserialization failure. This makes it
+   * possible to attribute "Failed to deserialize PubSubPosition" warnings to a specific store-version/partition
+   * instead of only the raw offset, which is otherwise insufficient to identify the affected replica.
+   *
+   * @param wireFormatBytes the serialized position bytes (can be null or empty)
+   * @param offset the fallback offset to use if deserialization fails or buffer is empty
+   * @param pubSubPositionDeserializer the deserializer to convert wire format to position
+   * @param replicaId canonical store-version/partition identity for logging context; may be {@code null} when the
+   *                  caller doesn't have replica context (e.g. off-server tooling), in which case "N/A" is logged
+   * @return a valid PubSubPosition, either deserialized or offset-based
+   */
+  public static PubSubPosition deserializePositionWithOffsetFallback(
+      ByteBuffer wireFormatBytes,
+      long offset,
+      PubSubPositionDeserializer pubSubPositionDeserializer,
+      String replicaId) {
     // Fast path: nothing to deserialize
     if (wireFormatBytes == null || !wireFormatBytes.hasRemaining()) {
       return fromKafkaOffset(offset);
@@ -387,7 +409,8 @@ public static PubSubPosition deserializePositionWithOffsetFallback(
           offset);
     } catch (RuntimeException e) {
       LOGGER.warn(
-          "Failed to deserialize PubSubPosition. Using offset-based position (offset={}, bufferRem={}, bufferCap={}).",
+          "Failed to deserialize PubSubPosition for replica: {}. Using offset-based position (offset={}, bufferRem={}, bufferCap={}).",
+          replicaId == null ? "N/A" : replicaId,
           offset,
           wireFormatBytes.remaining(),
           wireFormatBytes.capacity(),

internal/venice-common/src/test/java/com/linkedin/venice/pubsub/PubSubUtilTest.java

@@ -25,7 +25,17 @@
 import com.linkedin.venice.utils.ByteUtils;
 import com.linkedin.venice.utils.VeniceProperties;
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Properties;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.Appender;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.config.Configuration;
+import org.apache.logging.log4j.core.config.LoggerConfig;
+import org.apache.logging.log4j.core.config.Property;
 import org.testng.annotations.Test;
 
 
@@ -622,4 +632,60 @@ public void testDeserializePositionWithOffsetFallback() {
     actualPosition = PubSubUtil.deserializePositionWithOffsetFallback(zeroBuffer, 0L, deserializer);
     assertEquals(actualPosition.getNumericOffset(), 0L, "Should handle zero offset correctly");
   }
+
+  @Test
+  public void testDeserializePositionWithOffsetFallbackLogsReplicaIdOnDeserializationFailure() {
+    PubSubPositionDeserializer deserializer = PubSubPositionDeserializer.DEFAULT_DESERIALIZER;
+    // Malformed / truncated wire format bytes, mirroring the legacy malformed upstream position payloads
+    // that trigger the "Failed to deserialize PubSubPosition" warning in production.
+    ByteBuffer invalidBuffer = ByteBuffer.wrap(new byte[] { 0x01, 0x02, 0x03 });
+
+    List<String> capturedMessages = new ArrayList<>();
+    Appender appender =
+        new AbstractAppender("testDeserializePositionAppender", null, null, false, Property.EMPTY_ARRAY) {
+          @Override
+          public void append(LogEvent event) {
+            capturedMessages.add(event.getMessage().getFormattedMessage());
+          }
+        };
+    appender.start();
+
+    LoggerContext loggerContext = (LoggerContext) LogManager.getContext(false);
+    Configuration configuration = loggerContext.getConfiguration();
+    LoggerConfig loggerConfig = configuration.getLoggerConfig(PubSubUtil.class.getName());
+    loggerConfig.addAppender(appender, null, null);
+    loggerContext.updateLoggers();
+
+    try {
+      // Happy path for the new overload: replicaId is supplied and should be embedded in the warning
+      // so that the affected store-version/partition can be attributed without guesswork.
+      String replicaId = "cert1-histogram-hybrid_v43-7";
+      PubSubPosition positionWithReplicaId =
+          PubSubUtil.deserializePositionWithOffsetFallback(invalidBuffer, 42L, deserializer, replicaId);
+      assertEquals(
+          positionWithReplicaId.getNumericOffset(),
+          42L,
+          "Invalid buffer should still fall back to offset-based position");
+      assertTrue(
+          capturedMessages.stream().anyMatch(message -> message.contains(replicaId)),
+          "Warning log should include the supplied replicaId for debugging: " + capturedMessages);
+
+      // Edge case (R14): caller omits replica context (e.g. legacy 3-arg overload used by non-server callers).
+      // The warning must still be well-formed and clearly indicate the identity is unavailable ("N/A"),
+      // instead of throwing, logging null, or dropping the placeholder silently.
+      capturedMessages.clear();
+      PubSubPosition positionWithoutReplicaId =
+          PubSubUtil.deserializePositionWithOffsetFallback(invalidBuffer, 99L, deserializer);
+      assertEquals(
+          positionWithoutReplicaId.getNumericOffset(),
+          99L,
+          "Invalid buffer should still fall back to offset-based position");
+      assertTrue(
+          capturedMessages.stream().anyMatch(message -> message.contains("N/A")),
+          "Warning log should fall back to N/A when no replicaId context is available: " + capturedMessages);
+    } finally {
+      loggerConfig.removeAppender("testDeserializePositionAppender");
+      loggerContext.updateLoggers();
+    }
+  }
 }