[da-vinci][server] Global RT DIV: Metadata Partition (linkedin#2498)

Global RT DIV is stored in the metadata partition to avoid conflict with the user data. Added multiple new end-to-end integration tests.

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: KaiSernLim

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

@@ -30,8 +30,9 @@
 import com.linkedin.davinci.stats.ingestion.heartbeat.HeartbeatMonitoringService;
 import com.linkedin.davinci.storage.StorageService;
 import com.linkedin.davinci.storage.chunking.ChunkedValueManifestContainer;
-import com.linkedin.davinci.storage.chunking.GenericChunkingAdapter;
+import com.linkedin.davinci.storage.chunking.ChunkingUtils;
 import com.linkedin.davinci.storage.chunking.GenericRecordChunkingAdapter;
+import com.linkedin.davinci.storage.chunking.RawBytesChunkingAdapter;
 import com.linkedin.davinci.store.StorageEngine;
 import com.linkedin.davinci.store.StoragePartitionAdjustmentTrigger;
 import com.linkedin.davinci.store.cache.backend.ObjectCacheBackend;
@@ -112,6 +113,7 @@
 import it.unimi.dsi.fastutil.ints.Int2ObjectMap;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -177,7 +179,6 @@
  */
 public class LeaderFollowerStoreIngestionTask extends StoreIngestionTask {
   private static final Logger LOGGER = LogManager.getLogger(LeaderFollowerStoreIngestionTask.class);
-  public static final String GLOBAL_RT_DIV_KEY_PREFIX = "GLOBAL_RT_DIV_KEY.";
   static final long VIEW_WRITER_CLOSE_TIMEOUT_IN_MS = 60000; // 60s
 
   /**
@@ -3827,14 +3828,17 @@ private void produceToLocalKafkaHelper(
   }
 
   /**
-   * The Global RT DIV is produced on a per-broker basis, so the name includes the broker URL for differentiation.
+   * The Global RT DIV key includes the partition ID to prevent collisions when a server hosts multiple partitions,
+   * and the broker URL to distinguish between different upstream brokers.
    */
-  public static String getGlobalRtDivKeyName(String brokerUrl) {
-    return GLOBAL_RT_DIV_KEY_PREFIX + brokerUrl;
+  public static String getGlobalRtDivKeyName(int partitionId, String brokerUrl) {
+    return GLOBAL_RT_DIV_KEY_PREFIX + partitionId + "." + brokerUrl;
   }
 
-  public byte[] getGlobalRtDivKeyBytes(String brokerUrl) {
-    return globalRtDivKeyBytesCache.computeIfAbsent(brokerUrl, url -> getGlobalRtDivKeyName(url).getBytes());
+  public byte[] getGlobalRtDivKeyBytes(int partitionId, String brokerUrl) {
+    return globalRtDivKeyBytesCache.computeIfAbsent(
+        partitionId + "." + brokerUrl,
+        k -> getGlobalRtDivKeyName(partitionId, brokerUrl).getBytes(StandardCharsets.UTF_8));
   }
 
   /**
@@ -3852,7 +3856,7 @@ void sendGlobalRtDivMessage(
       long beforeProcessingRecordTimestampNs,
       LeaderMetadataWrapper leaderMetadataWrapper,
       LeaderProducedRecordContext context) {
-    final byte[] keyBytes = getGlobalRtDivKeyBytes(brokerUrl);
+    final byte[] keyBytes = getGlobalRtDivKeyBytes(partition, brokerUrl);
     final PubSubTopicPartition topicPartition = previousMessage.getTopicPartition();
     TopicType realTimeTopicType = TopicType.of(REALTIME_TOPIC_TYPE, brokerUrl);
 
@@ -3878,10 +3882,17 @@ void sendGlobalRtDivMessage(
         topicPartition,
         vtDiv);
 
-    // Get the old value manifest which contains the list of old chunks, so they can be deleted
-    final int schemaId = AvroProtocolDefinition.GLOBAL_RT_DIV_STATE.getCurrentProtocolVersion();
+    // Read the old manifest (if any) so VeniceWriter can delete orphaned old chunks in Kafka.
     ChunkedValueManifestContainer valueManifestContainer = new ChunkedValueManifestContainer();
-    readGlobalRtDivState(keyBytes, schemaId, topicPartition, valueManifestContainer);
+    byte[] manifestKey = ChunkingUtils.KEY_WITH_CHUNKING_SUFFIX_SERIALIZER.serializeNonChunkedKey(keyBytes);
+    byte[] oldRaw = storageEngine.getGlobalRtDivMetadata(manifestKey);
+    if (oldRaw != null && oldRaw.length >= ValueRecord.SCHEMA_HEADER_LENGTH) {
+      int schemaId = ValueRecord.parseSchemaId(oldRaw);
+      if (schemaId == AvroProtocolDefinition.CHUNKED_VALUE_MANIFEST.getCurrentProtocolVersion()) {
+        valueManifestContainer
+            .setManifest(ChunkingUtils.CHUNKED_VALUE_MANIFEST_SERIALIZER.deserialize(oldRaw, schemaId));
+      }
+    }
 
     // TODO: remove. this is a temporary log for debugging while the feature is in its infancy
     LOGGER.info(
@@ -3986,61 +3997,74 @@ private LeaderProducerCallback createGlobalRtDivCallback(
   }
 
   /**
-   * Reads a GlobalRtDivState value from the storage engine. Returns null if the value does not exist.
+   * Reads a GlobalRtDivState value from the metadata storage partition.
+   * Returns null if the key does not carry the expected prefix or the value does not exist.
+   * For chunked states, chunk assembly is performed at read time via {@link RawBytesChunkingAdapter}.
    *
-   * @param keyBytes the serialized key for the value
-   * @param readerValueSchemaID the schemaId to use for deserialization
+   * @param keyBytes the serialized key for the value (without any chunking suffix)
+   * @param readerValueSchemaID the schemaId to use when deserializing the assembled value
    * @param topicPartition the topic/partition for the value
-   * @param manifestContainer a container to store any manifest information retrieved from the storage engine
+   * @param manifestContainer populated with the {@link ChunkedValueManifest} if the value was chunked
    * @return the deserialized GlobalRtDivState object, or null if the value does not exist
    */
   GlobalRtDivState readGlobalRtDivState(
       byte[] keyBytes,
       int readerValueSchemaID,
       PubSubTopicPartition topicPartition,
       ChunkedValueManifestContainer manifestContainer) {
-    ByteBuffer valueBytes;
+    final String key = new String(keyBytes, StandardCharsets.UTF_8);
+    if (!key.startsWith(GLOBAL_RT_DIV_KEY_PREFIX)) {
+      return null;
+    }
+    final int partitionId = topicPartition.getPartitionNumber();
+    // Extract brokerUrl for logging (key format: GLOBAL_RT_DIV_KEY.{partitionId}.{brokerUrl})
+    final String afterPrefix = key.substring(GLOBAL_RT_DIV_KEY_PREFIX.length());
+    final int dotIdx = afterPrefix.indexOf('.');
+    final String brokerUrl = dotIdx >= 0 ? afterPrefix.substring(dotIdx + 1) : afterPrefix;
+
+    final BiFunction<Integer, ByteBuffer, byte[]> metadataGetter =
+        (part, keyBuf) -> storageEngine.getGlobalRtDivMetadata(ByteUtils.extractByteArray(keyBuf));
+    ByteBuffer assembledBytes;
     try {
-      valueBytes = (ByteBuffer) GenericChunkingAdapter.INSTANCE.get(
-          storageEngine,
-          topicPartition.getPartitionNumber(),
+      assembledBytes = RawBytesChunkingAdapter.INSTANCE.get(
+          metadataGetter,
+          storageEngine.getStoreVersionName(),
+          partitionId,
           ByteBuffer.wrap(keyBytes),
-          isChunked,
+          true, // must be true because Global RT DIV could always be large and require chunking
           null,
           null,
           NoOpReadResponseStats.SINGLETON,
           readerValueSchemaID,
           RawBytesStoreDeserializerCache.getInstance(),
           compressor.get(),
           manifestContainer);
-    } catch (Exception e) {
-      // TODO: evaluate whether these logs can be set to debug
+    } catch (VeniceException e) {
       LOGGER.error(
-          "Unable to retrieve the stored value bytes for key: {}, topic-partition: {}",
-          new String(keyBytes),
+          "Unable to read Global RT DIV state from metadata storage for topic-partition: {}, brokerUrl: {}",
           topicPartition,
+          brokerUrl,
           e);
       return null;
     }
-
-    if (valueBytes == null) {
-      // TODO: evaluate whether these logs can be set to debug
-      LOGGER.warn(
-          "No value found in the storage engine for key: {}, topic-partition: {}",
-          new String(keyBytes),
-          topicPartition);
+    if (assembledBytes == null) {
       return null;
     }
+    return deserializeGlobalRtDivState(ByteUtils.extractByteArray(assembledBytes), keyBytes, topicPartition);
+  }
 
+  private GlobalRtDivState deserializeGlobalRtDivState(
+      byte[] serializedValueBytes,
+      byte[] keyBytes,
+      PubSubTopicPartition topicPartition) {
     try {
-      return globalRtDivStateSerializer.deserialize(
-          ByteUtils.extractByteArray(valueBytes),
-          AvroProtocolDefinition.GLOBAL_RT_DIV_STATE.getCurrentProtocolVersion());
+      return globalRtDivStateSerializer
+          .deserialize(serializedValueBytes, AvroProtocolDefinition.GLOBAL_RT_DIV_STATE.getCurrentProtocolVersion());
     } catch (Exception e) {
       // TODO: evaluate whether these logs can be set to debug
       LOGGER.error(
           "Unable to deserialize stored value bytes for key: {}, topic-partition: {}",
-          new String(keyBytes),
+          new String(keyBytes, StandardCharsets.UTF_8),
           topicPartition,
           e);
       return null;
@@ -4147,14 +4171,13 @@ void loadGlobalRtDiv(int partition, String brokerUrl) {
     final PubSubTopic topic = pcs.getOffsetRecord().getLeaderTopic(getPubSubTopicRepository());
     final PubSubTopicPartition topicPartition = new PubSubTopicPartitionImpl(topic, pcs.getPartition());
 
-    String globalRtDivKey = getGlobalRtDivKeyName(brokerUrl);
-    byte[] keyBytes = globalRtDivKey.getBytes();
-    final ChunkedValueManifestContainer valueManifestContainer = new ChunkedValueManifestContainer();
+    byte[] keyBytes = getGlobalRtDivKeyBytes(partition, brokerUrl);
     GlobalRtDivState globalRtDivState = readGlobalRtDivState(
         keyBytes,
         AvroProtocolDefinition.GLOBAL_RT_DIV_STATE.getCurrentProtocolVersion(),
         topicPartition,
-        valueManifestContainer);
+        new ChunkedValueManifestContainer());
+
     if (globalRtDivState == null) {
       // If the GlobalRtDivState is not present, it could be acceptable if this could be the first leader to be elected
       // Object not existing could be problematic if this isn't the first leader (detected via nonzero leaderPosition)

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

@@ -223,6 +223,7 @@ public abstract class StoreIngestionTask implements Runnable, Closeable {
   protected static final int CHUNK_SCHEMA_ID = AvroProtocolDefinition.CHUNK.getCurrentProtocolVersion();
   private static final int CHUNK_MANIFEST_SCHEMA_ID =
       AvroProtocolDefinition.CHUNKED_VALUE_MANIFEST.getCurrentProtocolVersion();
+  public static final String GLOBAL_RT_DIV_KEY_PREFIX = "GLOBAL_RT_DIV_KEY.";
 
   protected static final RedundantExceptionFilter REDUNDANT_LOGGING_FILTER =
       RedundantExceptionFilter.getRedundantExceptionFilter();
@@ -4424,6 +4425,12 @@ protected void putInStorageEngine(int partition, byte[] keyBytes, Put put) {
     executeStorageEngineRunnable(partition, () -> storageEngine.put(partition, keyBytes, put.putValue));
   }
 
+  protected void putGlobalRtDivStateInMetadata(int partition, byte[] keyBytes, Put put) {
+    storageEngine.putGlobalRtDivMetadata(
+        keyBytes,
+        ByteUtils.prependIntHeaderToByteBuffer(put.putValue, put.schemaId, false).array());
+  }
+
   protected void removeFromStorageEngine(int partition, byte[] keyBytes, Delete delete) {
     executeStorageEngineRunnable(partition, () -> storageEngine.delete(partition, keyBytes));
   }
@@ -4626,7 +4633,9 @@ private int processKafkaDataMessage(
 
         int writerSchemaId = put.getSchemaId();
 
-        if (recordTransformer != null && messageType == MessageType.PUT) {
+        if (kafkaKey.isGlobalRtDiv()) {
+          putGlobalRtDivStateInMetadata(producedPartition, keyBytes, put);
+        } else if (recordTransformer != null && messageType == MessageType.PUT) {
           long recordTransformerStartTime = System.nanoTime();
           ByteBufferValueRecord<ByteBuffer> assembledRecord = chunkAssembler.bufferAndAssembleRecord(
               consumerRecord.getTopicPartition(),
@@ -4792,7 +4801,11 @@ private int processKafkaDataMessage(
         }
 
         keyLen = keyBytes.length;
-        deleteFromStorageEngine(producedPartition, keyBytes, delete);
+        if (kafkaKey.isGlobalRtDiv()) {
+          storageEngine.deleteGlobalRtDivMetadata(keyBytes);
+        } else {
+          deleteFromStorageEngine(producedPartition, keyBytes, delete);
+        }
         if (recordMetrics) {
           double deleteLatency = LatencyUtils.getElapsedTimeFromNSToMS(startTimeNs);
           versionedIngestionStats.recordStorageEngineDeleteTime(storeName, versionNumber, deleteLatency);

clients/da-vinci-client/src/main/java/com/linkedin/davinci/storage/chunking/AbstractAvroChunkingAdapter.java

@@ -16,6 +16,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.ByteBuffer;
+import java.util.function.BiFunction;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.io.BinaryDecoder;
 
@@ -135,6 +136,42 @@ public T get(
         manifestContainer);
   }
 
+  /**
+   * Variant of {@link #get} that accepts a custom storage getter function instead of a {@link StorageEngine}.
+   * Use this when the data lives outside a regular data partition (e.g., in a metadata partition) and a
+   * custom lookup is needed to locate both the manifest and the individual chunks.
+   */
+  public T get(
+      BiFunction<Integer, ByteBuffer, byte[]> storageGetter,
+      String storeVersionName,
+      int partition,
+      ByteBuffer key,
+      boolean isChunked,
+      T reusedValue,
+      BinaryDecoder reusedDecoder,
+      ReadResponseStats responseStats,
+      int readerSchemaId,
+      StoreDeserializerCache<T> storeDeserializerCache,
+      VeniceCompressor compressor,
+      ChunkedValueManifestContainer manifestContainer) {
+    if (isChunked) {
+      key = ChunkingUtils.KEY_WITH_CHUNKING_SUFFIX_SERIALIZER.serializeNonChunkedKey(key);
+    }
+    return ChunkingUtils.getFromStorage(
+        this,
+        (p, k) -> storageGetter.apply(p, k),
+        storeVersionName,
+        partition,
+        key,
+        responseStats,
+        reusedValue,
+        reusedDecoder,
+        readerSchemaId,
+        storeDeserializerCache,
+        compressor,
+        manifestContainer);
+  }
+
   public ByteBufferValueRecord<T> getWithSchemaId(
       StorageEngine store,
       int partition,

clients/da-vinci-client/src/main/java/com/linkedin/davinci/storage/chunking/ChunkingUtils.java

@@ -61,7 +61,7 @@
  *    a chunked value.
  */
 public class ChunkingUtils {
-  static final ChunkedValueManifestSerializer CHUNKED_VALUE_MANIFEST_SERIALIZER =
+  public static final ChunkedValueManifestSerializer CHUNKED_VALUE_MANIFEST_SERIALIZER =
       new ChunkedValueManifestSerializer(false);
   public static final KeyWithChunkingSuffixSerializer KEY_WITH_CHUNKING_SUFFIX_SERIALIZER =
       new KeyWithChunkingSuffixSerializer();

clients/da-vinci-client/src/main/java/com/linkedin/davinci/store/AbstractStorageEngine.java

@@ -669,6 +669,30 @@ public synchronized void clearStoreVersionState() {
     metadataPartition.delete(VERSION_METADATA_KEY);
   }
 
+  @Override
+  public synchronized void putGlobalRtDivMetadata(byte[] keyBytes, byte[] valueWithHeader) {
+    if (!metadataPartitionCreated()) {
+      throw new StorageInitializationException("Metadata partition not created!");
+    }
+    metadataPartition.put(keyBytes, valueWithHeader);
+  }
+
+  @Override
+  public synchronized byte[] getGlobalRtDivMetadata(byte[] keyBytes) {
+    if (!metadataPartitionCreated()) {
+      return null;
+    }
+    return metadataPartition.get(keyBytes);
+  }
+
+  @Override
+  public synchronized void deleteGlobalRtDivMetadata(byte[] keyBytes) {
+    if (!metadataPartitionCreated()) {
+      return;
+    }
+    metadataPartition.delete(keyBytes);
+  }
+
   /**
    * Return true or false based on whether a given partition exists within this storage engine
    *

clients/da-vinci-client/src/main/java/com/linkedin/davinci/store/DelegatingStorageEngine.java

@@ -279,6 +279,21 @@ public void clearStoreVersionState() {
     this.delegate.clearStoreVersionState();
   }
 
+  @Override
+  public void putGlobalRtDivMetadata(byte[] keyBytes, byte[] valueWithHeader) {
+    this.delegate.putGlobalRtDivMetadata(keyBytes, valueWithHeader);
+  }
+
+  @Override
+  public byte[] getGlobalRtDivMetadata(byte[] keyBytes) {
+    return this.delegate.getGlobalRtDivMetadata(keyBytes);
+  }
+
+  @Override
+  public void deleteGlobalRtDivMetadata(byte[] keyBytes) {
+    this.delegate.deleteGlobalRtDivMetadata(keyBytes);
+  }
+
   @Override
   public boolean containsPartition(int partitionId) {
     return this.delegate.containsPartition(partitionId);

clients/da-vinci-client/src/main/java/com/linkedin/davinci/store/StorageEngine.java

@@ -12,6 +12,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.Supplier;
+import javax.annotation.Nullable;
 
 
 public interface StorageEngine<Partition extends AbstractStoragePartition> extends Closeable {
@@ -154,6 +155,27 @@ void putWithReplicationMetadata(int partitionId, byte[] key, ByteBuffer value, b
    */
   void clearStoreVersionState();
 
+  /**
+   * Store a raw GlobalRtDiv entry (chunk, manifest, or full value — all with schema header prepended)
+   * directly in the metadata partition at the given raw key bytes.
+   * The key must already contain the VeniceWriter chunking suffix so that
+   * {@link com.linkedin.davinci.storage.chunking.GenericChunkingAdapter} can assemble it at read time.
+   */
+  void putGlobalRtDivMetadata(byte[] keyBytes, byte[] valueWithHeader);
+
+  /**
+   * Retrieve a raw GlobalRtDiv entry from the metadata partition at the given raw key bytes.
+   *
+   * @return the stored bytes (schema header prepended), or {@code null} if not present
+   */
+  @Nullable
+  byte[] getGlobalRtDivMetadata(byte[] keyBytes);
+
+  /**
+   * Delete a raw GlobalRtDiv entry from the metadata partition at the given raw key bytes.
+   */
+  void deleteGlobalRtDivMetadata(byte[] keyBytes);
+
   /**
    * Return true or false based on whether a given partition exists within this storage engine
    *