[server] Switch back to PubSubPosition based reads with offset as fallback #2242

sushantmane · 2025-10-28T20:08:54Z

Switch back to PubSubPosition based reads with offset as fallback

Code changes

Added new code behind a config. If so list the config names and their default values in the PR description.
Introduced new log lines.
- Confirmed if logs need to be rate limited to avoid excessive logging.

Concurrency-Specific Checks

Both reviewer and PR author to verify

Code has no race conditions or thread safety issues.
Proper synchronization mechanisms (e.g., synchronized, RWLock) are used where needed.
No blocking calls inside critical sections that could lead to deadlocks or performance degradation.
Verified thread-safe collections are used (e.g., ConcurrentHashMap, CopyOnWriteArrayList).
Validated proper exception handling in multi-threaded code to avoid silent thread termination.

How was this PR tested?

New unit tests added.
New integration tests added.
Modified or extended existing tests.
Verified backward compatibility (if applicable).

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 removes the getPubSubPositionString method and its associated test, replacing its usage with direct calls to pubSubPositionDeserializer.toPosition(). The key changes enable better position comparison by using actual PubSubPosition objects rather than numeric offsets, and utilize the deserializePositionWithOffsetFallback method for safer deserialization with fallback logic.

Removes PubSubUtil.getPubSubPositionString utility method and its test
Replaces numeric offset comparisons with proper position comparisons using diffPosition and object equality
Updates deserializePositionWithOffsetFallback to remove the offset > 0 guard condition
Integrates proper position deserialization with fallback in extractUpstreamPosition

Reviewed Changes

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

Show a summary per file

File	Description
PubSubUtil.java	Removes `getPubSubPositionString` utility method
PubSubUtilTest.java	Removes test coverage for `getPubSubPositionString`
OffsetRecord.java	Removes unused import and updates `deserializePositionWithOffsetFallback` condition
KafkaTopicDumper.java	Replaces `getPubSubPositionString` with direct `toPosition` calls
StoreIngestionTask.java	Updates position deserialization condition and integrates fallback in `extractUpstreamPosition`
StoreIngestionTaskTest.java	Changes test assertions to use full position equality instead of numeric offset comparison
PartitionTracker.java	Replaces `getPubSubPositionString` with direct `toPosition` call
LeaderFollowerStoreIngestionTask.java	Replaces numeric offset comparisons with proper position comparisons using `diffPosition` and symbolic position checks

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

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

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

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

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

github-actions · 2025-11-28T02:10:45Z

Hi there. This pull request has been inactive for 30 days. To keep our review queue healthy, we plan to close it in 7 days unless there is new activity. If you are still working on this, please push a commit, leave a comment, or convert it to draft to signal intent. Thank you for your time and contributions.

Copilot

Pull request overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated 2 comments.

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

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

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

…lback Remove additional safeguards

…bolic positions The deserializePositionWithOffsetFallback method was incorrectly treating symbolic positions (EARLIEST, LATEST) as regular positions and comparing their numeric offsets against the provided minimum offset. This caused EARLIEST (numeric offset -1) to be replaced with an offset-based position when the minimum offset was >= 0. Added a guard to detect and preserve symbolic positions before performing numeric offset comparison, ensuring they are returned as-is regardless of the minimum offset parameter. This fix resolves the failing testDeserializePositionWithOffsetFallback test in LeaderFollowerStoreIngestionTaskTest. Test: ./gradlew :clients:da-vinci-client:test --tests "com.linkedin.davinci.kafka.consumer.LeaderFollowerStoreIngestionTaskTest"

Copilot

Pull request overview

Copilot reviewed 15 out of 15 changed files in this pull request and generated 2 comments.

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

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

KaiSernLim · 2025-12-18T02:41:54Z

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

  }
-
-  @Test
-  public void testGetPubSubPositionString() {


I think PubSubUtil.getPubSubPositionString() should also be removed at this point?

KaiSernLim · 2025-12-18T03:16:14Z

internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java

+   * causes issues in production.
+   */
+  public static final String SERVER_USE_UPSTREAM_PUBSUB_POSITION_WITH_FALLBACK =
+      "server.use.upstream.pubsub.position.with.fallback";


Suggested change

"server.use.upstream.pubsub.position.with.fallback";

"server.use.upstream.pubsub.positions";

I think it's okay to leave out the fallback part in the config and rely on it being mentioned in the docstring. How about a shorter name like this? If not, I feel like it should be possible to arrive at something shorter than what's in the PR.

Same with checkpointed.

KaiSernLim · 2025-12-18T06:58:01Z

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

+  public static PubSubPosition deserializePositionWithOffsetFallback(
+      ByteBuffer wireFormatBytes,
+      long offset,
+      PubSubPositionDeserializer pubSubPositionDeserializer) {
+    // Fast path: nothing to deserialize
+    if (wireFormatBytes == null || !wireFormatBytes.hasRemaining()) {
+      return fromKafkaOffset(offset);
+    }
+
+    try {
+      final PubSubPosition position = pubSubPositionDeserializer.toPosition(wireFormatBytes);
+
+      // Guard against regressions: honor the caller-provided minimum offset.
+      // This applies to both symbolic and concrete positions.
+      if (position.getNumericOffset() < offset) {
+        LOGGER.info(
+            "Deserialized position: {} is behind the provided offset: {}. Using offset-based position.",
+            position.getNumericOffset(),
+            offset);
+        return fromKafkaOffset(offset);
+      }
+
+      // If position is ahead of or equal to offset, return it as-is (including symbolic positions like LATEST)
+      return position;
+    } catch (RuntimeException e) {
+      LOGGER.warn(
+          "Failed to deserialize PubSubPosition. Using offset-based position (offset={}, bufferRem={}, bufferCap={}).",
+          offset,
+          wireFormatBytes.remaining(),
+          wireFormatBytes.capacity(),
+          e);
+      return fromKafkaOffset(offset);
+    }
+  }


Suggested change

public static PubSubPosition deserializePositionWithOffsetFallback(

ByteBuffer wireFormatBytes,

long offset,

PubSubPositionDeserializer pubSubPositionDeserializer) {

// Fast path: nothing to deserialize

if (wireFormatBytes == null || !wireFormatBytes.hasRemaining()) {

return fromKafkaOffset(offset);

}

try {

final PubSubPosition position = pubSubPositionDeserializer.toPosition(wireFormatBytes);

// Guard against regressions: honor the caller-provided minimum offset.

// This applies to both symbolic and concrete positions.

if (position.getNumericOffset() < offset) {

LOGGER.info(

"Deserialized position: {} is behind the provided offset: {}. Using offset-based position.",

position.getNumericOffset(),

offset);

return fromKafkaOffset(offset);

}

// If position is ahead of or equal to offset, return it as-is (including symbolic positions like LATEST)

return position;

} catch (RuntimeException e) {

LOGGER.warn(

"Failed to deserialize PubSubPosition. Using offset-based position (offset={}, bufferRem={}, bufferCap={}).",

offset,

wireFormatBytes.remaining(),

wireFormatBytes.capacity(),

e);

return fromKafkaOffset(offset);

}

}

public static PubSubPosition deserializePositionWithOffsetFallback(

ByteBuffer wireFormatBytes,

long offset,

PubSubPositionDeserializer pubSubPositionDeserializer) {

if (wireFormatBytes != null && wireFormatBytes.hasRemaining()) {

try {

final PubSubPosition position = pubSubPositionDeserializer.toPosition(wireFormatBytes);

if (position.getNumericOffset() >= offset) {

return position; // Valid position: ahead of or equal to offset

}

LOGGER.info(

"Deserialized position: {} is behind the provided offset: {}. Using offset-based position.",

position.getNumericOffset(),

offset);

} catch (RuntimeException e) {

LOGGER.warn(

"Failed to deserialize PubSubPosition. Using offset-based position (offset={}, bufferRem={}, bufferCap={}).",

offset,

wireFormatBytes.remaining(),

wireFormatBytes.capacity(),

e);

}

}

// Offset fallback for all invalid or missing cases

return fromKafkaOffset(offset);

}

I see every branch having the same offset fallback return value. How about this layout, instead?

It's a bit less-explicit with the individual fallback cases, but I find it a bit more clear to have one single offset fallback case at the end.

KaiSernLim · 2025-12-18T08:06:10Z

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

    return pubSubContext;
  }

  PubSubPosition deserializePositionWithOffsetFallback(


Isn't this just basically the same as the PubSubUtil version? Can it be unified?

KaiSernLim · 2025-12-18T08:09:14Z

...ient/src/main/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTask.java

+        if ((PubSubSymbolicPosition.EARLIEST.equals(latestConsumedRtPosition)
+            && !PubSubSymbolicPosition.EARLIEST.equals(upstreamStartPosition))


I don't quite understand what the first condition means

KaiSernLim · 2025-12-18T08:11:12Z

...da-vinci-client/src/main/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerImpl.java

-                  .stream()
-                  .map(bb -> pubSubPositionDeserializer.toPosition(bb).getNumericOffset())
-                  .collect(Collectors.toList());
+          List<Long> highWatermarkOffsets;


Suggested change

List<Long> highWatermarkOffsets;

List<ByteBuffer> positions = versionSwap.getLocalHighWatermarkPubSubPositions();

List<Long> offsets = versionSwap.getLocalHighWatermarks();

List<Long> highWatermarkOffsets = new ArrayList<>();

if (positions != null && !positions.isEmpty()) {

for (int i = 0; i < positions.size(); i++) {

long fallbackOffset = (offsets != null && i < offsets.size()) ? offsets.get(i) : -1L;

PubSubPosition position = PubSubUtil

.deserializePositionWithOffsetFallback(positions.get(i), fallbackOffset, pubSubPositionDeserializer);

highWatermarkOffsets.add(position.getNumericOffset());

}

}

nit: how about this?

Copilot AI review requested due to automatic review settings October 28, 2025 20:08

Copilot AI reviewed Oct 28, 2025

View reviewed changes

github-actions bot added the stale label Nov 28, 2025

sushantmane force-pushed the sumane/switch-back-reads-to-PubSubPosition branch from 6fa7342 to e86b854 Compare December 3, 2025 19:38

Copilot AI review requested due to automatic review settings December 3, 2025 22:47

sushantmane force-pushed the sumane/switch-back-reads-to-PubSubPosition branch from e86b854 to cd66555 Compare December 3, 2025 22:47

Copilot started reviewing on behalf of sushantmane December 3, 2025 22:48 View session

Copilot finished reviewing on behalf of sushantmane December 3, 2025 22:52

Copilot AI reviewed Dec 3, 2025

View reviewed changes

internal/venice-common/src/main/java/com/linkedin/venice/offsets/OffsetRecord.java Outdated Show resolved Hide resolved

clients/venice-admin-tool/src/main/java/com/linkedin/venice/KafkaTopicDumper.java Outdated Show resolved Hide resolved

github-actions bot removed the stale label Dec 4, 2025

sushantmane and others added 4 commits December 9, 2025 14:23

[server] Swtich back to PubSubPosition based reads with offset as fal…

39ee410

…lback Remove additional safeguards

Move deserializePositionWithOffsetFallback to PubSubUtil

46f5241

Add fallback

4d32025

sushantmane force-pushed the sumane/switch-back-reads-to-PubSubPosition branch from cd66555 to a5271ff Compare December 9, 2025 22:26

sushantmane added 2 commits December 9, 2025 14:55

Commit all changes

5e2005e

Fix test failures

a80f0a5

sushantmane force-pushed the sumane/switch-back-reads-to-PubSubPosition branch from a5271ff to a80f0a5 Compare December 9, 2025 22:58

Copilot AI review requested due to automatic review settings December 9, 2025 22:58

Copilot started reviewing on behalf of sushantmane December 9, 2025 22:58 View session

Copilot AI reviewed Dec 9, 2025

View reviewed changes

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

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

KaiSernLim reviewed Dec 18, 2025

View reviewed changes

	"server.use.upstream.pubsub.position.with.fallback";
	"server.use.upstream.pubsub.positions";

		if ((PubSubSymbolicPosition.EARLIEST.equals(latestConsumedRtPosition)
		&& !PubSubSymbolicPosition.EARLIEST.equals(upstreamStartPosition))

-          List<Long> highWatermarkOffsets;
+          List<ByteBuffer> positions = versionSwap.getLocalHighWatermarkPubSubPositions();
+          List<Long> offsets = versionSwap.getLocalHighWatermarks();
+          List<Long> highWatermarkOffsets = new ArrayList<>();
+          if (positions != null && !positions.isEmpty()) {
+            for (int i = 0; i < positions.size(); i++) {
+              long fallbackOffset = (offsets != null && i < offsets.size()) ? offsets.get(i) : -1L;
+              PubSubPosition position = PubSubUtil
+                  .deserializePositionWithOffsetFallback(positions.get(i), fallbackOffset, pubSubPositionDeserializer);
+              highWatermarkOffsets.add(position.getNumericOffset());
+            }
+          }

[server] Switch back to PubSubPosition based reads with offset as fallback #2242

Are you sure you want to change the base?

[server] Switch back to PubSubPosition based reads with offset as fallback #2242

Uh oh!

Conversation

sushantmane commented Oct 28, 2025

Switch back to PubSubPosition based reads with offset as fallback

Code changes

Concurrency-Specific Checks

How was this PR tested?

Does this PR introduce any user-facing or breaking changes?

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

github-actions bot commented Nov 28, 2025

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

KaiSernLim Dec 18, 2025

Choose a reason for hiding this comment

Uh oh!

KaiSernLim Dec 18, 2025

Choose a reason for hiding this comment

Uh oh!

KaiSernLim Dec 18, 2025

Choose a reason for hiding this comment

Uh oh!

KaiSernLim Dec 18, 2025

Choose a reason for hiding this comment

Uh oh!

KaiSernLim Dec 18, 2025

Choose a reason for hiding this comment

Uh oh!

KaiSernLim Dec 18, 2025

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants