Address review comments

Author: sushantmane

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

@@ -97,7 +97,6 @@
 import com.linkedin.venice.meta.Version;
 import com.linkedin.venice.offsets.OffsetRecord;
 import com.linkedin.venice.pubsub.PubSubContext;
-import com.linkedin.venice.pubsub.PubSubPositionDeserializer;
 import com.linkedin.venice.pubsub.PubSubTopicPartitionImpl;
 import com.linkedin.venice.pubsub.PubSubTopicRepository;
 import com.linkedin.venice.pubsub.PubSubUtil;
@@ -4890,11 +4889,10 @@ protected PubSubPosition extractUpstreamPosition(DefaultPubSubMessage consumerRe
 
     // Check config to determine whether to use upstreamPubSubPosition with fallback or just upstreamOffset
     if (storeVersionConfig.isUseUpstreamPubSubPositionWithFallbackEnabled()) {
-      return deserializePositionWithOffsetFallback(
-          pubSubContext.getPubSubPositionDeserializer(),
-          consumerRecord.getTopicPartition(),
+      return PubSubUtil.deserializePositionWithOffsetFallback(
           leaderMetadataFooter.upstreamPubSubPosition,
-          leaderMetadataFooter.upstreamOffset);
+          leaderMetadataFooter.upstreamOffset,
+          pubSubContext.getPubSubPositionDeserializer());
     } else {
       // Directly use upstreamOffset without attempting position deserialization
       return PubSubUtil.fromKafkaOffset(leaderMetadataFooter.upstreamOffset);
@@ -5204,55 +5202,6 @@ PubSubContext getPubSubContext() {
     return pubSubContext;
   }
 
-  PubSubPosition deserializePositionWithOffsetFallback(
-      PubSubPositionDeserializer pubSubPositionDeserializer,
-      PubSubTopicPartition topicPartition,
-      ByteBuffer wireFormatBytes,
-      long offset) {
-    // Fast path: nothing to deserialize
-    if (wireFormatBytes == null || !wireFormatBytes.hasRemaining()) {
-      return PubSubUtil.fromKafkaOffset(offset);
-    }
-
-    try {
-      final PubSubPosition position = pubSubPositionDeserializer.toPosition(wireFormatBytes);
-
-      // Symbolic positions (EARLIEST, LATEST) should always be returned as-is
-      if (position == PubSubSymbolicPosition.EARLIEST || position == PubSubSymbolicPosition.LATEST) {
-        return position;
-      }
-
-      // Guard against regressions: honor the caller-provided minimum offset.
-      if (position.getNumericOffset() < offset) {
-        String context = String.format(" for: %s/%s", topicPartition, versionTopic);
-        if (!REDUNDANT_LOGGING_FILTER.isRedundantException(context)) {
-          LOGGER.warn(
-              "Deserialized position: {} is behind the provided offset: {}. Using offset-based position for: {}/{}",
-              position,
-              offset,
-              topicPartition,
-              versionTopic);
-        }
-        return PubSubUtil.fromKafkaOffset(offset);
-      }
-
-      return position;
-    } catch (Exception e) {
-      String context = String.format("%s/%s - %s", topicPartition, versionTopic, e.getMessage());
-      if (!REDUNDANT_LOGGING_FILTER.isRedundantException(context)) {
-        LOGGER.warn(
-            "Failed to deserialize PubSubPosition for: {}/{}. Using offset-based position (offset={}, bufferRem={}, bufferCap={}).",
-            topicPartition,
-            versionTopic,
-            offset,
-            wireFormatBytes.remaining(),
-            wireFormatBytes.capacity(),
-            e);
-      }
-      return PubSubUtil.fromKafkaOffset(offset);
-    }
-  }
-
   /**
    * Validates that END_OF_PUSH has been received before processing TOPIC_SWITCH.
    *

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

@@ -65,7 +65,6 @@
 import com.linkedin.venice.offsets.OffsetRecord;
 import com.linkedin.venice.partitioner.DefaultVenicePartitioner;
 import com.linkedin.venice.pubsub.PubSubContext;
-import com.linkedin.venice.pubsub.PubSubPositionDeserializer;
 import com.linkedin.venice.pubsub.PubSubTopicImpl;
 import com.linkedin.venice.pubsub.PubSubTopicPartitionImpl;
 import com.linkedin.venice.pubsub.PubSubTopicRepository;
@@ -867,47 +866,6 @@ public void testIsLocalVersionTopicPartitionFullyConsumed(
     }
   }
 
-  @Test(timeOut = 60_000)
-  public void testDeserializePositionWithOffsetFallback() throws InterruptedException {
-    setUp();
-
-    // Use DEFAULT_DESERIALIZER instead of mocking
-    PubSubPositionDeserializer deserializer = PubSubPositionDeserializer.DEFAULT_DESERIALIZER;
-    PubSubTopicPartition mockTopicPartition = mock(PubSubTopicPartition.class);
-    when(mockTopicPartition.toString()).thenReturn("test-topic_v1-0");
-
-    // Use doCallRealMethod to test the actual implementation
-    doCallRealMethod().when(leaderFollowerStoreIngestionTask)
-        .deserializePositionWithOffsetFallback(any(), any(), any(), anyLong());
-    // Case 1: Null ByteBuffer - should return offset-based position
-    PubSubPosition actualPosition = leaderFollowerStoreIngestionTask
-        .deserializePositionWithOffsetFallback(deserializer, mockTopicPartition, null, 100L);
-    assertEquals(actualPosition.getNumericOffset(), 100L, "Null ByteBuffer should return offset-based position");
-
-    // Case 2: Empty ByteBuffer - should return offset-based position
-    actualPosition = leaderFollowerStoreIngestionTask
-        .deserializePositionWithOffsetFallback(deserializer, mockTopicPartition, ByteBuffer.allocate(0), 200L);
-    assertEquals(actualPosition.getNumericOffset(), 200L, "Empty ByteBuffer should return offset-based position");
-
-    // Case 3: Invalid ByteBuffer - should return offset-based position
-    ByteBuffer invalidBuffer = ByteBuffer.wrap(new byte[] { 0x01, 0x02, 0x03 }); // Random invalid bytes
-    actualPosition = leaderFollowerStoreIngestionTask
-        .deserializePositionWithOffsetFallback(deserializer, mockTopicPartition, invalidBuffer, 300L);
-    assertEquals(actualPosition.getNumericOffset(), 300L, "Invalid ByteBuffer should return offset-based position");
-
-    // Case 4: Valid ByteBuffer - should return deserialized position
-    ByteBuffer validBuffer = ApacheKafkaOffsetPosition.of(400L).toWireFormatBuffer();
-    actualPosition = leaderFollowerStoreIngestionTask
-        .deserializePositionWithOffsetFallback(deserializer, mockTopicPartition, validBuffer, 0L);
-    assertEquals(actualPosition.getNumericOffset(), 400L, "Valid ByteBuffer should return deserialized position");
-
-    // Case 5: ByteBuffer has EARLIEST symbolic position and offset is 0
-    ByteBuffer earliestBuffer = PubSubSymbolicPosition.EARLIEST.toWireFormatBuffer();
-    actualPosition = leaderFollowerStoreIngestionTask
-        .deserializePositionWithOffsetFallback(deserializer, mockTopicPartition, earliestBuffer, 0L);
-    assertEquals(actualPosition, PubSubSymbolicPosition.EARLIEST, "Should return EARLIEST symbolic position");
-  }
-
   @Test(timeOut = 60_000)
   public void testExtractUpstreamClusterId() throws InterruptedException {
     setUp();

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

@@ -374,6 +374,11 @@ public static PubSubPosition deserializePositionWithOffsetFallback(
 
     try {
       PubSubPosition position = pubSubPositionDeserializer.toPosition(wireFormatBytes);
+      // Symbolic positions (EARLIEST, LATEST) should always be returned as-is
+      if (position == PubSubSymbolicPosition.EARLIEST || position == PubSubSymbolicPosition.LATEST) {
+        return position;
+      }
+
       // Guard against regressions: honor the caller-provided minimum offset.
       // This applies to both symbolic and concrete positions.
       if (position.getNumericOffset() >= offset) {

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

@@ -563,4 +563,63 @@ public void testParsePositionWireFormatInvalidInputs() {
         () -> PubSubUtil.parsePositionWireFormat("1:invalid_base64!", deserializer));
     expectThrows(IllegalArgumentException.class, () -> PubSubUtil.parsePositionWireFormat("1:", deserializer));
   }
+
+  @Test
+  public void testDeserializePositionWithOffsetFallback() {
+    PubSubPositionDeserializer deserializer = PubSubPositionDeserializer.DEFAULT_DESERIALIZER;
+
+    // Case 1: Null ByteBuffer - should return offset-based position
+    PubSubPosition actualPosition = PubSubUtil.deserializePositionWithOffsetFallback(null, 100L, deserializer);
+    assertEquals(actualPosition.getNumericOffset(), 100L, "Null ByteBuffer should return offset-based position");
+
+    // Case 2: Empty ByteBuffer - should return offset-based position
+    actualPosition = PubSubUtil.deserializePositionWithOffsetFallback(ByteBuffer.allocate(0), 200L, deserializer);
+    assertEquals(actualPosition.getNumericOffset(), 200L, "Empty ByteBuffer should return offset-based position");
+
+    // Case 3: Invalid ByteBuffer - should return offset-based position
+    ByteBuffer invalidBuffer = ByteBuffer.wrap(new byte[] { 0x01, 0x02, 0x03 }); // Random invalid bytes
+    actualPosition = PubSubUtil.deserializePositionWithOffsetFallback(invalidBuffer, 300L, deserializer);
+    assertEquals(actualPosition.getNumericOffset(), 300L, "Invalid ByteBuffer should return offset-based position");
+
+    // Case 4: Valid ByteBuffer with position ahead of offset - should return deserialized position
+    ByteBuffer validBuffer = ApacheKafkaOffsetPosition.of(400L).toWireFormatBuffer();
+    actualPosition = PubSubUtil.deserializePositionWithOffsetFallback(validBuffer, 0L, deserializer);
+    assertEquals(actualPosition.getNumericOffset(), 400L, "Valid ByteBuffer should return deserialized position");
+
+    // Case 5: Valid ByteBuffer with position equal to offset - should return deserialized position
+    ByteBuffer equalBuffer = ApacheKafkaOffsetPosition.of(500L).toWireFormatBuffer();
+    actualPosition = PubSubUtil.deserializePositionWithOffsetFallback(equalBuffer, 500L, deserializer);
+    assertEquals(
+        actualPosition.getNumericOffset(),
+        500L,
+        "Position equal to offset should return deserialized position");
+
+    // Case 6: Valid ByteBuffer with position behind offset - should return offset-based position
+    ByteBuffer behindBuffer = ApacheKafkaOffsetPosition.of(100L).toWireFormatBuffer();
+    actualPosition = PubSubUtil.deserializePositionWithOffsetFallback(behindBuffer, 200L, deserializer);
+    assertEquals(actualPosition.getNumericOffset(), 200L, "Position behind offset should return offset-based position");
+
+    // Case 7: ByteBuffer has EARLIEST symbolic position - should return EARLIEST as-is
+    ByteBuffer earliestBuffer = PubSubSymbolicPosition.EARLIEST.toWireFormatBuffer();
+    actualPosition = PubSubUtil.deserializePositionWithOffsetFallback(earliestBuffer, 100L, deserializer);
+    assertEquals(actualPosition, PubSubSymbolicPosition.EARLIEST, "Should return EARLIEST symbolic position as-is");
+
+    // Case 8: ByteBuffer has LATEST symbolic position - should return LATEST as-is
+    ByteBuffer latestBuffer = PubSubSymbolicPosition.LATEST.toWireFormatBuffer();
+    actualPosition = PubSubUtil.deserializePositionWithOffsetFallback(latestBuffer, 100L, deserializer);
+    assertEquals(actualPosition, PubSubSymbolicPosition.LATEST, "Should return LATEST symbolic position as-is");
+
+    // Case 9: Large offset values
+    ByteBuffer largeBuffer = ApacheKafkaOffsetPosition.of(Long.MAX_VALUE - 1000L).toWireFormatBuffer();
+    actualPosition = PubSubUtil.deserializePositionWithOffsetFallback(largeBuffer, 0L, deserializer);
+    assertEquals(
+        actualPosition.getNumericOffset(),
+        Long.MAX_VALUE - 1000L,
+        "Should handle large offset values correctly");
+
+    // Case 10: Zero offset
+    ByteBuffer zeroBuffer = ApacheKafkaOffsetPosition.of(0L).toWireFormatBuffer();
+    actualPosition = PubSubUtil.deserializePositionWithOffsetFallback(zeroBuffer, 0L, deserializer);
+    assertEquals(actualPosition.getNumericOffset(), 0L, "Should handle zero offset correctly");
+  }
 }