diff --git a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/BlobTransferIngestionHelper.java b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/BlobTransferIngestionHelper.java index bbab0c383f1..9c1f66c1967 100644 --- a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/BlobTransferIngestionHelper.java +++ b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/BlobTransferIngestionHelper.java @@ -138,6 +138,7 @@ public boolean isReplicaLaggedAndNeedBlobTransfer( LOGGER.warn("Offset record not found for: {}", replicaId); return true; } + offsetRecord.setReplicaId(replicaId); // TODO: Remove offset lag threshold entirely in the future — no offset lag should be allowed. long blobTransferDisabledOffsetLagThreshold = serverConfig.getBlobTransferDisabledOffsetLagThreshold(); if (blobTransferDisabledOffsetLagThreshold < 0) { diff --git a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/PartitionConsumptionState.java b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/PartitionConsumptionState.java index 75a5f8a916f..ea9e2db9449 100644 --- a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/PartitionConsumptionState.java +++ b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/PartitionConsumptionState.java @@ -426,6 +426,7 @@ public PartitionConsumptionState( } this.hybrid = hybrid; this.offsetRecord = offsetRecord; + this.offsetRecord.setReplicaId(getReplicaId()); this.pubSubContext = pubSubContext; this.errorReported = false; this.lagCaughtUp = false; diff --git a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java index 90d8780c4f6..946d240700c 100644 --- a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java +++ b/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); diff --git a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/validation/PartitionTracker.java b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/validation/PartitionTracker.java index de8c171f0ac..52bf24d8885 100644 --- a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/validation/PartitionTracker.java +++ b/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: ") diff --git a/clients/venice-admin-tool/src/main/java/com/linkedin/venice/KafkaTopicDumper.java b/clients/venice-admin-tool/src/main/java/com/linkedin/venice/KafkaTopicDumper.java index c55894112cd..0c53c2fa9ce 100644 --- a/clients/venice-admin-tool/src/main/java/com/linkedin/venice/KafkaTopicDumper.java +++ b/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); } diff --git a/clients/venice-push-job/src/main/java/com/linkedin/venice/spark/consistency/LilyPadSnapshotBuilder.java b/clients/venice-push-job/src/main/java/com/linkedin/venice/spark/consistency/LilyPadSnapshotBuilder.java index 51b26aa2216..bc60cee138d 100644 --- a/clients/venice-push-job/src/main/java/com/linkedin/venice/spark/consistency/LilyPadSnapshotBuilder.java +++ b/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 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) { diff --git a/internal/venice-common/src/main/java/com/linkedin/venice/offsets/OffsetRecord.java b/internal/venice-common/src/main/java/com/linkedin/venice/offsets/OffsetRecord.java index a6cb56bd038..fc690d8ebf6 100644 --- a/internal/venice-common/src/main/java/com/linkedin/venice/offsets/OffsetRecord.java +++ b/internal/venice-common/src/main/java/com/linkedin/venice/offsets/OffsetRecord.java @@ -54,6 +54,11 @@ public class OffsetRecord { private final PubSubContext pubSubContext; private final PubSubPositionDeserializer pubSubPositionDeserializer; + /** + * Runtime-only diagnostic context. This is intentionally kept outside {@link PartitionState}, so binding an + * OffsetRecord to a replica does not change its serialized form. + */ + private volatile String replicaId; private PubSubTopic leaderPubSubTopic; public OffsetRecord( @@ -136,7 +141,8 @@ public void clearPreviousStatusesEntry(CharSequence key) { public PubSubPosition getCheckpointedLocalVtPosition() { return deserializePositionWithOffsetFallback( this.partitionState.lastProcessedVersionTopicPubSubPosition, - this.partitionState.offset); + this.partitionState.offset, + positionSourceOrDefault("OffsetRecord.lastProcessedVersionTopicPubSubPosition")); } public void checkpointLocalVtPosition(PubSubPosition vtPosition) { @@ -151,7 +157,8 @@ public void checkpointLocalVtPosition(PubSubPosition vtPosition) { public PubSubPosition getCheckpointedRemoteVtPosition() { return deserializePositionWithOffsetFallback( this.partitionState.upstreamVersionTopicPubSubPosition, - this.partitionState.upstreamVersionTopicOffset); + this.partitionState.upstreamVersionTopicOffset, + positionSourceOrDefault("OffsetRecord.upstreamVersionTopicPubSubPosition")); } public void checkpointRemoteVtPosition(PubSubPosition remoteVtPosition) { @@ -311,7 +318,10 @@ public PubSubPosition getCheckpointedRtPosition(String pubSubBrokerAddress) { // If the offset is not set, return EARLIEST symbolic position. return PubSubSymbolicPosition.EARLIEST; } - return deserializePositionWithOffsetFallback(wfBuffer, offset); + return deserializePositionWithOffsetFallback( + wfBuffer, + offset, + positionSourceOrDefault("OffsetRecord.upstreamRealTimeTopicPubSubPosition[" + pubSubBrokerAddress + "]")); } public void checkpointRtPosition(String pubSubBrokerAddress, PubSubPosition leaderPosition) { @@ -341,12 +351,36 @@ public void cloneRtPositionCheckpoints(@Nonnull Map chec for (Map.Entry 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( + "OffsetRecord.upstreamRealTimeTopicPubSubPosition[" + pubSubBrokerAddress + "]"))); } } } + /** + * Associates runtime diagnostic context with this record. Rebinding to the same replica is harmless, while + * rebinding to a different replica is rejected because an OffsetRecord must not be shared across partitions. + * + *

The synchronized write and volatile read make the binding safely visible to threads that subsequently + * deserialize checkpointed positions.

+ */ + public synchronized void setReplicaId(String replicaId) { + if (replicaId == null || replicaId.isEmpty()) { + throw new IllegalArgumentException("Replica ID cannot be null or empty"); + } + if (this.replicaId != null && !this.replicaId.equals(replicaId)) { + throw new IllegalStateException( + "OffsetRecord is already associated with replica " + this.replicaId + " and cannot be rebound to " + + replicaId); + } + this.replicaId = replicaId; + } + public GUID getLeaderGUID() { return this.partitionState.leaderGUID; } @@ -551,10 +585,29 @@ 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); + } + + private String positionSourceOrDefault(String fallbackLabel) { + String currentReplicaId = replicaId; + return currentReplicaId != null ? currentReplicaId : fallbackLabel; } } diff --git a/internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubUtil.java b/internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubUtil.java index ab171ca8a08..1e39171cae9 100644 --- a/internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubUtil.java +++ b/internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubUtil.java @@ -362,11 +362,52 @@ public static PubSubPositionGrpcWireFormat getPubSubPositionGrpcWireFormat(PubSu * @param offset the fallback offset to use if deserialization fails or buffer is empty * @param pubSubPositionDeserializer the deserializer to convert wire format to position * @return a valid PubSubPosition, either deserialized or offset-based + * @deprecated Prefer {@link #deserializePositionWithOffsetFallback(ByteBuffer, long, PubSubPositionDeserializer, String)} + * so that a failure warning can be attributed to its source. This overload is retained only for tests and + * truly context-free callers; it logs {@code "N/A"} as the position source. */ + @Deprecated 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 positionSource} that is included in the warning logged on deserialization failure. This makes + * it possible to attribute "Failed to deserialize PubSubPosition" warnings to a specific origin instead of only + * the raw offset, which is otherwise insufficient to identify what emitted the malformed payload. + * + *

{@code positionSource} is intentionally a free-form string rather than a strongly-typed replica handle, + * because not every caller has a replica (store-version/partition) to point to. Callers should supply the most + * specific stable identifier they have, in this order of preference: + *

    + *
  1. a canonical replica ID, e.g. {@code _v-} as produced by + * {@code Utils.getReplicaId}, when ingesting/consuming a specific partition
    2. + *
  2. a broker/field-qualified label (e.g. {@code "OffsetRecord.upstreamRealTimeTopicPubSubPosition[]"}) + * when the caller only has a checkpoint field and no partition context
    3. + *
  3. a static, descriptive call-site label (e.g. {@code "OffsetRecord.lastProcessedVersionTopicPubSubPosition"}) + * for utility code that has no per-call context at all
    4. + *
+ * Fabricating an identifier that doesn't reflect real context (e.g. reusing an unrelated ID) defeats the purpose + * and must be avoided; passing {@code null} is only acceptable when no caller-supplied context is possible, in + * which case {@code "N/A"} is logged. + * + * @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 positionSource a stable, non-fabricated diagnostic label identifying the origin of {@code wireFormatBytes} + * for logging context; may be {@code null} when no context is available, 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 positionSource) { // Fast path: nothing to deserialize if (wireFormatBytes == null || !wireFormatBytes.hasRemaining()) { return fromKafkaOffset(offset); @@ -387,7 +428,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: {}. Using offset-based position (offset={}, bufferRem={}, bufferCap={}).", + positionSource == null ? "N/A" : positionSource, offset, wireFormatBytes.remaining(), wireFormatBytes.capacity(), diff --git a/internal/venice-common/src/test/java/com/linkedin/venice/offsets/TestOffsetRecord.java b/internal/venice-common/src/test/java/com/linkedin/venice/offsets/TestOffsetRecord.java index 4c7c4d5dec4..fc080261853 100644 --- a/internal/venice-common/src/test/java/com/linkedin/venice/offsets/TestOffsetRecord.java +++ b/internal/venice-common/src/test/java/com/linkedin/venice/offsets/TestOffsetRecord.java @@ -8,6 +8,7 @@ import com.linkedin.venice.kafka.protocol.GUID; import com.linkedin.venice.kafka.protocol.state.PartitionState; import com.linkedin.venice.kafka.protocol.state.ProducerPartitionState; +import com.linkedin.venice.pubsub.PubSubUtil; import com.linkedin.venice.pubsub.adapter.kafka.common.ApacheKafkaOffsetPosition; import com.linkedin.venice.pubsub.api.PubSubPosition; import com.linkedin.venice.pubsub.api.PubSubSymbolicPosition; @@ -15,11 +16,21 @@ import com.linkedin.venice.utils.TestUtils; import com.linkedin.venice.utils.concurrent.VeniceConcurrentHashMap; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Optional; import java.util.function.Supplier; +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.Assert; import org.testng.annotations.BeforeClass; import org.testng.annotations.DataProvider; @@ -111,6 +122,111 @@ public void testResetUpstreamOffsetMap() { Assert.assertEquals(offsetRecord.getCheckpointedRtPosition(TEST_KAFKA_URL2), p2); } + /** + * R14 edge/failure case: when the wire-format bytes for a checkpointed position are malformed, + * {@link OffsetRecord} must never let the deserialization-failure warning attribute the failure to a bare + * {@code null}/"N/A". It should use its bound replicaId or, before it is associated with a replica, fall back to a + * stable field-scoped label that identifies which checkpoint accessor/field is affected. + */ + @Test + public void testCheckpointedPositionsLogNonNullSourceOnDeserializationFailure() { + ByteBuffer malformedBuffer = ByteBuffer.wrap(new byte[] { 0x01, 0x02, 0x03 }); + PartitionState corruptState = new PartitionState(); + corruptState.offset = 777L; + corruptState.lastProcessedVersionTopicPubSubPosition = malformedBuffer.duplicate(); + corruptState.upstreamOffsetMap = new VeniceConcurrentHashMap<>(); + corruptState.upstreamOffsetMap.put(TEST_KAFKA_URL1, 888L); + corruptState.upstreamRealTimeTopicPubSubPositionMap = new VeniceConcurrentHashMap<>(); + corruptState.upstreamRealTimeTopicPubSubPositionMap.put(TEST_KAFKA_URL1, malformedBuffer.duplicate()); + OffsetRecord corruptRecord = new OffsetRecord( + corruptState, + AvroProtocolDefinition.PARTITION_STATE.getSerializer(), + DEFAULT_PUBSUB_CONTEXT_FOR_UNIT_TESTING); + + List capturedMessages = new ArrayList<>(); + Appender appender = new AbstractAppender( + "testCheckpointedPositionsLogNonNullSourceAppender", + 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 { + // Before replica association, use a field-scoped label rather than null/"N/A". + assertEquals(corruptRecord.getCheckpointedLocalVtPosition().getNumericOffset(), 777L); + assertTrue( + capturedMessages.stream() + .anyMatch( + message -> message.contains("Failed to deserialize PubSubPosition") + && message.contains("OffsetRecord.lastProcessedVersionTopicPubSubPosition")), + "Expected field-scoped fallback label in: " + capturedMessages); + assertTrue( + capturedMessages.stream() + .noneMatch( + message -> message.contains("Failed to deserialize PubSubPosition") && message.contains("N/A")), + "Should not need the N/A sentinel when a field-scoped label is available: " + capturedMessages); + + capturedMessages.clear(); + assertEquals(corruptRecord.getCheckpointedRtPosition(TEST_KAFKA_URL1).getNumericOffset(), 888L); + assertTrue( + capturedMessages.stream() + .anyMatch( + message -> message.contains("Failed to deserialize PubSubPosition") + && message.contains("OffsetRecord.upstreamRealTimeTopicPubSubPosition[" + TEST_KAFKA_URL1 + "]")), + "Expected broker-qualified fallback label in: " + capturedMessages); + + String replicaId = "myStore_v3-5"; + corruptRecord.setReplicaId(replicaId); + capturedMessages.clear(); + assertEquals(corruptRecord.getCheckpointedLocalVtPosition().getNumericOffset(), 777L); + assertTrue( + capturedMessages.stream() + .anyMatch( + message -> message.contains("Failed to deserialize PubSubPosition") && message.contains(replicaId)), + "Expected bound replicaId in: " + capturedMessages); + + capturedMessages.clear(); + assertEquals(corruptRecord.getCheckpointedRtPosition(TEST_KAFKA_URL1).getNumericOffset(), 888L); + assertTrue( + capturedMessages.stream() + .anyMatch( + message -> message.contains("Failed to deserialize PubSubPosition") && message.contains(replicaId)), + "Expected bound replicaId in: " + capturedMessages); + } finally { + loggerConfig.removeAppender("testCheckpointedPositionsLogNonNullSourceAppender"); + loggerContext.updateLoggers(); + appender.stop(); + } + } + + @Test + public void testReplicaIdBindingIsRuntimeOnlyAndCannotChange() { + OffsetRecord record = new OffsetRecord( + AvroProtocolDefinition.PARTITION_STATE.getSerializer(), + DEFAULT_PUBSUB_CONTEXT_FOR_UNIT_TESTING); + byte[] serializedBeforeBinding = record.toBytes(); + + record.setReplicaId("myStore_v3-5"); + record.setReplicaId("myStore_v3-5"); + + assertTrue( + Arrays.equals(serializedBeforeBinding, record.toBytes()), + "Runtime replica identity must not alter serialized PartitionState"); + Assert.expectThrows(IllegalStateException.class, () -> record.setReplicaId("anotherStore_v1-0")); + } + @Test public void testBatchUpdateEOIP() { OffsetRecord offsetRecord = TestUtils diff --git a/internal/venice-common/src/test/java/com/linkedin/venice/pubsub/PubSubUtilTest.java b/internal/venice-common/src/test/java/com/linkedin/venice/pubsub/PubSubUtilTest.java index 50867c9d543..841bcb61bf6 100644 --- a/internal/venice-common/src/test/java/com/linkedin/venice/pubsub/PubSubUtilTest.java +++ b/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,80 @@ 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 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: a canonical 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("Failed to deserialize PubSubPosition") && message.contains(replicaId)), + "Warning log should include the supplied replicaId for debugging: " + capturedMessages); + + // positionSource is intentionally free-form: callers without a replica handle (e.g. OffsetRecord's + // checkpoint accessors) may instead supply a static, field-scoped label. That must be logged verbatim too. + capturedMessages.clear(); + String fieldLabel = "OffsetRecord.upstreamRealTimeTopicPubSubPosition[dc-0_kafka]"; + PubSubPosition positionWithFieldLabel = + PubSubUtil.deserializePositionWithOffsetFallback(invalidBuffer, 55L, deserializer, fieldLabel); + assertEquals( + positionWithFieldLabel.getNumericOffset(), + 55L, + "Invalid buffer should still fall back to offset-based position"); + assertTrue( + capturedMessages.stream() + .anyMatch( + message -> message.contains("Failed to deserialize PubSubPosition") && message.contains(fieldLabel)), + "Warning log should include the supplied field-scoped label for debugging: " + capturedMessages); + + // Edge case (R14): caller omits all context (e.g. legacy 3-arg overload retained only for tests/truly + // context-free 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 positionWithoutContext = + PubSubUtil.deserializePositionWithOffsetFallback(invalidBuffer, 99L, deserializer); + assertEquals( + positionWithoutContext.getNumericOffset(), + 99L, + "Invalid buffer should still fall back to offset-based position"); + assertTrue( + capturedMessages.stream() + .anyMatch(message -> message.contains("Failed to deserialize PubSubPosition") && message.contains("N/A")), + "Warning log should fall back to N/A when no context at all is available: " + capturedMessages); + } finally { + loggerConfig.removeAppender("testDeserializePositionAppender"); + loggerContext.updateLoggers(); + appender.stop(); + } + } }