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

[sushantmane]

Problem Statement

Venice server 0.15.182 emitted repeated Failed to deserialize PubSubPosition. Using offset-based position... warnings. The deployed logging omitted which replica (store-version/partition) triggered the warning, so we could only approximate the affected stores by correlating nearby log lines — attribution was weak and not reliable enough to act on.

Root cause: a legacy malformed 10-byte upstreamPubSubPosition wire-format payload in the leader metadata footer. PubSubUtil#deserializePositionWithOffsetFallback correctly and safely falls back to the numeric offset, but the warning it logs carried no context beyond the raw offset.

The first commit on this PR added a replicaId-aware overload and wired it through the single hottest call site (StoreIngestionTask#extractUpstreamPosition), but left every other call site (OffsetRecord, KafkaTopicDumper, LilyPadSnapshotBuilder, PartitionTracker) on the legacy overload, which logs "N/A". Review feedback asked: can we log real context for all paths instead of null?

Solution

Inventoried every production call site of PubSubUtil#deserializePositionWithOffsetFallback and gave each one the best context it can honestly provide, instead of fabricating identifiers:

Call site	Context now supplied
StoreIngestionTask#extractUpstreamPosition	canonical replicaId (already fixed)
PartitionTracker (DIV violation message)	Utils.getReplicaId(consumerRecord.getTopicPartition()) — the violating record's own topic-partition, previously unused here
KafkaTopicDumper (per-record metadata log, TopicSwitch log)	Utils.getReplicaId(record.getTopicPartition()) — the admin tool already has the topic-partition of the record being dumped
LilyPadSnapshotBuilder	Utils.getReplicaId(iterator.getTopicPartition()) — same pattern
OffsetRecord's 4 checkpoint accessors (getCheckpointedLocalVtPosition, getCheckpointedRemoteVtPosition, getCheckpointedRtPosition, cloneRtPositionCheckpoints)	see below

OffsetRecord itself has no notion of which store-version/partition it belongs to — that identity lives in the caller (e.g. PartitionConsumptionState). Fabricating a replicaId inside OffsetRecord would be dishonest, so instead:

• Added a String replicaId overload of each accessor so callers that do have replica context (PartitionConsumptionState's constructor and getLatestProcessedRtPosition, IngestionNotificationDispatcher, BlobTransferIngestionHelper, and StoreIngestionTask's remaining 4 call sites) thread it through instead of silently passing null.
• For the handful of genuinely context-free paths that remain (OffsetRecord#toString/toSimplifiedString, DeepCopyOffsetManager, InMemoryOffsetManager, tests), the accessors fall back to a stable, static, field-scoped label (e.g. "OffsetRecord.lastProcessedVersionTopicPubSubPosition" or "OffsetRecord.upstreamRealTimeTopicPubSubPosition[<broker>]") instead of null, so the warning still identifies which checkpoint field is affected even without a replica identity.

PubSubUtil#deserializePositionWithOffsetFallback's 4th parameter is renamed from replicaId to the more accurate positionSource, since not every caller has a true replica handle to offer. Its Javadoc documents the preferred ordering of context callers should supply (replicaId > broker/field-qualified label > static call-site label) and explicitly calls out that fabricating an unrelated identifier is worse than the null/"N/A" fallback. The 3-arg overload is kept and marked @Deprecated, retained only for PubSubUtilTest and any future truly context-free callers — no production call site uses it anymore.

Code changes

• Introduced new log lines (this only enriches existing, already-rate-limited/occasional warnings; it does not add a new logging call site or increase log volume).
  • Confirmed no additional rate limiting is required.

How was this PR tested?

• New unit tests added.

• New integration tests added.

• Modified or extended existing tests.

• Verified backward compatibility (if applicable).

• Extended PubSubUtilTest's deserialization-failure logging test with a case asserting a non-replica field-scoped label is logged verbatim, alongside the existing replicaId and N/A-fallback cases.

• Added TestOffsetRecord#testCheckpointedPositionsLogNonNullSourceOnDeserializationFailure (R14 edge/failure case with malformed wire-format bytes): attaches a Log4j2 appender to PubSubUtil's logger and asserts the local-VT/RT checkpoint accessors log the field-scoped (broker-qualified, for RT) label when called without a replicaId, and log the supplied replicaId when given one — never null/"N/A".

• Updated 3 existing shouldStartBlobTransfer* mocks in LeaderFollowerStoreIngestionTaskTest to stub the new getCheckpointedLocalVtPosition(String) overload, since BlobTransferIngestionHelper now threads its replicaId through.

Ran locally (all green):

./gradlew :internal:venice-common:test \
    --tests "com.linkedin.venice.pubsub.PubSubUtilTest" \
    --tests "com.linkedin.venice.offsets.TestOffsetRecord"

./gradlew :clients:da-vinci-client:test \
    --tests "com.linkedin.davinci.validation.TestPartitionTracker" \
    --tests "com.linkedin.davinci.kafka.consumer.PartitionConsumptionStateTest" \
    --tests "com.linkedin.davinci.kafka.consumer.IngestionNotificationDispatcherTest" \
    --tests "com.linkedin.davinci.kafka.consumer.LeaderFollowerStoreIngestionTaskTest" \
    --tests "com.linkedin.davinci.kafka.consumer.ActiveActiveStoreIngestionTaskTest" \
    --tests "com.linkedin.davinci.kafka.consumer.PushTimeoutTest" \
    --tests "com.linkedin.davinci.store.rocksdb.RocksDBStorageEngineTest" \
    --tests "com.linkedin.davinci.kafka.consumer.SITWithPWiseAndBufferAfterLeaderTest" \
    --tests "com.linkedin.davinci.kafka.consumer.SITWithSAwarePWiseWithoutBufferAfterLeaderTest" \
    --tests "com.linkedin.davinci.kafka.consumer.SITWithPWiseWithoutBufferAfterLeaderTest" \
    --tests "com.linkedin.davinci.kafka.consumer.SITWithSAwarePWiseAndBufferAfterLeaderTest"

./gradlew :clients:venice-admin-tool:test --tests "com.linkedin.venice.TestKafkaTopicDumper"
./gradlew :clients:venice-push-job:test --tests "com.linkedin.venice.spark.consistency.LilyPadSnapshotBuilderTest"

./gradlew :internal:venice-common:compileJava :clients:da-vinci-client:compileJava \
    :clients:venice-admin-tool:compileJava :clients:venice-push-job:compileJava

./gradlew spotlessJavaApply  # no additional formatting changes needed

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

• No. You can skip the rest of this section.

[Copilot]

Pull request overview

This PR enriches the existing PubSubUtil.deserializePositionWithOffsetFallback deserialization-failure warning by including a replica identifier (<store>_v<version>-<partition>), improving debuggability of production warnings without changing fallback behavior.

Changes:

• Added a new deserializePositionWithOffsetFallback(..., String replicaId) overload that logs replica context (or N/A when absent).
• Kept the existing 3-arg overload by delegating to the new overload with replicaId = null.
• Wired replicaId through the ingestion hot path (StoreIngestionTask#extractUpstreamPosition) and added a unit test to validate the logging behavior.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.

File	Description
internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubUtil.java	Introduces the new overload and adds replicaId (N/A fallback) to the warning message on deserialization failure.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java	Passes canonical replicaId into the new overload at the ingestion call site responsible for the observed warnings.
internal/venice-common/src/test/java/com/linkedin/venice/pubsub/PubSubUtilTest.java	Adds a unit test that attaches an appender and asserts the warning includes replicaId (or N/A for the legacy overload).

---

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

[Copilot]

Pull request overview

Copilot reviewed 12 out of 12 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (10)

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

• The test appender is started but never stopped. Even though it is removed from the LoggerConfig, leaving it started can leak resources across the JVM and make other tests harder to debug.

    } finally {
      loggerConfig.removeAppender("testDeserializePositionAppender");
      loggerContext.updateLoggers();
    }

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

• Same flakiness risk as above: the appender may capture unrelated log messages if the resolved LoggerConfig is a parent/root. Restrict the match to the expected PubSubUtil warning before asserting on fieldLabel presence.

      assertTrue(
          capturedMessages.stream().anyMatch(message -> message.contains(fieldLabel)),
          "Warning log should include the supplied field-scoped label for debugging: " + capturedMessages);

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

• To avoid false positives from unrelated log lines (since the appender may be attached to a parent/root LoggerConfig), restrict the check for "N/A" to the specific PubSubUtil deserialization-failure warning.

      assertTrue(
          capturedMessages.stream().anyMatch(message -> message.contains("N/A")),
          "Warning log should fall back to N/A when no context at all is available: " + capturedMessages);

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

• For the same reason as the label assertion above, scope the "N/A" check to PubSubUtil's deserialization-failure warning; otherwise unrelated logs could cause a false failure.

      assertTrue(
          capturedMessages.stream().noneMatch(message -> message.contains("N/A")),
          "Should not need the N/A sentinel when a field-scoped label is available: " + capturedMessages);

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

• To reduce the chance of matching an unrelated log line when the appender is attached to a parent/root LoggerConfig, restrict this predicate to PubSubUtil's "Failed to deserialize PubSubPosition" warning.

      assertTrue(
          capturedMessages.stream().anyMatch(message -> message.contains(replicaId)),
          "Expected replicaId in: " + capturedMessages);

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

• Same as above: if the appender ends up attached to a parent/root logger, it can capture unrelated messages. Restrict the predicate to PubSubUtil's deserialization-failure warning before checking for the broker-qualified label.

      assertTrue(
          capturedMessages.stream()
              .anyMatch(
                  message -> message
                      .contains("OffsetRecord.upstreamRealTimeTopicPubSubPosition[" + TEST_KAFKA_URL1 + "]")),
          "Expected broker-qualified fallback label in: " + capturedMessages);

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

• Same flakiness risk: scope the replicaId match to PubSubUtil's deserialization-failure warning so unrelated logs can't satisfy the predicate.

      assertTrue(
          capturedMessages.stream().anyMatch(message -> message.contains(replicaId)),
          "Expected replicaId in: " + capturedMessages);

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

• The log-capture appender is attached to whatever LoggerConfig Log4j2 resolves for PubSubUtil (often the root logger). These assertions can become flaky if unrelated logs are captured during the test run. Filter to the specific "Failed to deserialize PubSubPosition" warning before checking for the replicaId.

This issue also appears in the following locations of the same file:

• line 683
• line 697

      assertTrue(
          capturedMessages.stream().anyMatch(message -> message.contains(replicaId)),
          "Warning log should include the supplied replicaId for debugging: " + capturedMessages);

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

• This test attaches an appender via Configuration#getLoggerConfig(PubSubUtil), which may resolve to a parent/root LoggerConfig and capture unrelated log messages. Restrict the match to the expected PubSubUtil deserialization-failure warning to avoid flakiness.

This issue also appears in the following locations of the same file:

• line 174
• line 182
• line 190
• line 199

      assertTrue(
          capturedMessages.stream()
              .anyMatch(message -> message.contains("OffsetRecord.lastProcessedVersionTopicPubSubPosition")),
          "Expected field-scoped fallback label in: " + capturedMessages);

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

• The test appender is started but never stopped. Stopping it in the finally block helps prevent leaking appenders/resources across the JVM and keeps other tests isolated.

    } finally {
      loggerConfig.removeAppender("testCheckpointedPositionsLogNonNullSourceAppender");
      loggerContext.updateLoggers();
    }

[Copilot]

Pull request overview

Copilot reviewed 10 out of 10 changed files in this pull request and generated no new comments.

Suppressed comments (2)

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

• In cloneRtPositionCheckpoints, the broker-qualified label string is concatenated unconditionally inside the loop ("..." + pubSubBrokerAddress + "]"), even when replicaId is set and will be used instead. This creates unnecessary allocations for every entry in upstreamOffsetMap. Cache the current replicaId once and only build the fallback label when it’s actually needed.

            deserializePositionWithOffsetFallback(
                wfBuffer,
                offsetEntry.getValue(),
                positionSourceOrDefault(
                    "OffsetRecord.upstreamRealTimeTopicPubSubPosition[" + pubSubBrokerAddress + "]")));

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

• positionSourceOrDefault("OffsetRecord.upstreamRealTimeTopicPubSubPosition[" + pubSubBrokerAddress + "]") eagerly builds the broker-qualified label even when replicaId is already bound, because the string concatenation happens before the method call. This adds avoidable allocation on a potentially hot path (RT checkpoint reads) and the label is discarded when replicaId != null. Build the label lazily only when replicaId is absent.

This issue also appears on line 356 of the same file.

    return deserializePositionWithOffsetFallback(
        wfBuffer,
        offset,
        positionSourceOrDefault("OffsetRecord.upstreamRealTimeTopicPubSubPosition[" + pubSubBrokerAddress + "]"));
  }