[common] Fall back to producer timestamp when pub-sub system timestamp is

unavailable

Venice currently relies on the pub-sub system (broker) timestamp for
getPubSubMessageTime(), but this timestamp is not always available or
accurate across all pub-sub systems. Venice already embeds a reliable
producer timestamp inside every KafkaMessageEnvelope via
producerMetadata.messageTimestamp, which is always set by VeniceWriter.

This change modifies PubSubMessageDeserializer to prefer the pub-sub
system timestamp when it is available and non-zero, but fall back to the
Venice producer timestamp when the pub-sub system provides no per-message
timestamp (timestamp is null or 0).

This is a no-op for existing Kafka deployments with LOG_APPEND_TIME since
broker timestamps are always valid and non-zero. For pub-sub systems that
do not provide reliable per-message timestamps, this ensures all 35+
downstream consumers of getPubSubMessageTime() — including DIV validation,
latency metrics, heartbeat processing, and CDC events — receive a
meaningful timestamp without any individual code changes.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: haoxu07

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

@@ -30,7 +30,8 @@ default POSITION getOffset() {
   POSITION getPosition();
 
   /**
-   * @return the timestamp at which the message was persisted in the pub sub system
+   * @return the best-available message timestamp. This is the pub-sub system timestamp when available and
+   *         non-zero, otherwise the Venice producer timestamp embedded in the message envelope.
    */
   long getPubSubMessageTime();
 

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

@@ -81,13 +81,17 @@ public DefaultPubSubMessage deserialize(
     if (value == null) {
       value = valueSerializer.deserialize(valueBytes, getEnvelope(key.getKeyHeaderByte()));
     }
+    // Prefer Venice's own producer timestamp when the pub-sub system timestamp is missing or zero.
+    // Some pub-sub systems do not provide reliable per-message timestamps. Venice always embeds a
+    // producer timestamp in the KafkaMessageEnvelope, so we can fall back to it.
+    long effectiveTimestamp = timestamp != null && timestamp > 0 ? timestamp : value.producerMetadata.messageTimestamp;
     // TODO: Put the message container in an object pool as well
     return new ImmutablePubSubMessage(
         key,
         value,
         topicPartition,
         pubSubPosition,
-        timestamp,
+        effectiveTimestamp,
         keyBytes.length + valueBytes.length,
         headers);
   }

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

@@ -132,6 +132,52 @@ public void testDeserializerValueWithInvalidSchemaFromPubSubMessageHeaders() {
     assertEquals(message.getPosition(), position);
   }
 
+  @Test
+  public void testDeserializerFallsBackToProducerTimestampWhenBrokerTimestampIsZero() {
+    KafkaKey key = new KafkaKey(MessageType.PUT, "key".getBytes());
+    KafkaMessageEnvelope value = getDummyValue();
+    long producerTimestamp = 1700000000000L;
+    value.producerMetadata.messageTimestamp = producerTimestamp;
+
+    // When broker timestamp is 0, should fall back to producer timestamp
+    DefaultPubSubMessage message = messageDeserializer.deserialize(
+        topicPartition,
+        keySerializer.serialize("test", key),
+        valueSerializer.serialize("test", value),
+        new PubSubMessageHeaders(),
+        position,
+        0L);
+    assertEquals(message.getPubSubMessageTime(), producerTimestamp);
+
+    // When broker timestamp is valid, should use broker timestamp
+    long brokerTimestamp = 1700000001000L;
+    message = messageDeserializer.deserialize(
+        topicPartition,
+        keySerializer.serialize("test", key),
+        valueSerializer.serialize("test", value),
+        new PubSubMessageHeaders(),
+        position,
+        brokerTimestamp);
+    assertEquals(message.getPubSubMessageTime(), brokerTimestamp);
+  }
+
+  @Test
+  public void testDeserializerFallsBackToProducerTimestampWhenBrokerTimestampIsNull() {
+    KafkaKey key = new KafkaKey(MessageType.PUT, "key".getBytes());
+    KafkaMessageEnvelope value = getDummyValue();
+    long producerTimestamp = 1700000000000L;
+    value.producerMetadata.messageTimestamp = producerTimestamp;
+
+    DefaultPubSubMessage message = messageDeserializer.deserialize(
+        topicPartition,
+        keySerializer.serialize("test", key),
+        valueSerializer.serialize("test", value),
+        new PubSubMessageHeaders(),
+        position,
+        null);
+    assertEquals(message.getPubSubMessageTime(), producerTimestamp);
+  }
+
   private KafkaMessageEnvelope getDummyValue() {
     KafkaMessageEnvelope value = new KafkaMessageEnvelope();
     value.producerMetadata = new ProducerMetadata();