🤖 Address Copilot review comments: input validation and nullability docs

- Add key-length guard before Arrays.copyOf in chunked GlobalRtDiv manifest
  path to avoid NegativeArraySizeException on corrupted/short keys
- Add METADATA_PARTITION_ID check to putGlobalRtDivChunk/getGlobalRtDivChunk
  for consistency with other metadata partition APIs
- Annotate getGlobalRtDivChunk return type with @nullable and add Javadoc
  to make the null-when-missing contract explicit at the interface level

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

Author: KaiSernLim

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

@@ -4121,6 +4121,10 @@ protected void putGlobalRtDivStateInMetadata(int partition, byte[] keyBytes, Put
     if (put.schemaId == CHUNK_MANIFEST_SCHEMA_ID) {
       // Manifest for a chunked GlobalRtDiv: assemble all chunks from the metadata partition,
       // decompress the result, then persist it in the metadata partition.
+      if (keyBytes.length < KEY_CHUNKING_SUFFIX_LENGTH) {
+        throw new VeniceException(
+            "GlobalRtDiv chunk manifest key too short to contain chunking suffix: " + Arrays.toString(keyBytes));
+      }
       byte[] originalKeyBytes = Arrays.copyOf(keyBytes, keyBytes.length - KEY_CHUNKING_SUFFIX_LENGTH);
       String key = new String(originalKeyBytes);
       if (!key.startsWith(GLOBAL_RT_DIV_KEY_PREFIX)) {

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

@@ -727,6 +727,10 @@ public synchronized void putGlobalRtDivChunk(int partitionId, byte[] chunkKey, b
     if (!metadataPartitionCreated()) {
       throw new StorageInitializationException("Metadata partition not created!");
     }
+    if (partitionId == METADATA_PARTITION_ID) {
+      throw new IllegalArgumentException(
+          "Metadata partition id should not be used as argument in putGlobalRtDivChunk.");
+    }
     if (partitionId < 0) {
       throw new IllegalArgumentException("Invalid partition id argument in putGlobalRtDivChunk");
     }
@@ -738,6 +742,10 @@ public synchronized byte[] getGlobalRtDivChunk(int partitionId, byte[] chunkKey)
     if (!metadataPartitionCreated()) {
       throw new StorageInitializationException("Metadata partition not created!");
     }
+    if (partitionId == METADATA_PARTITION_ID) {
+      throw new IllegalArgumentException(
+          "Metadata partition id should not be used as argument in getGlobalRtDivChunk.");
+    }
     if (partitionId < 0) {
       throw new IllegalArgumentException("Invalid partition id argument in getGlobalRtDivChunk");
     }

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 {
@@ -176,7 +177,10 @@ void putWithReplicationMetadata(int partitionId, byte[] key, ByteBuffer value, b
 
   /**
    * Retrieve a GlobalRtDiv intermediate chunk from the metadata partition.
+   *
+   * @return the chunk bytes, or {@code null} if the chunk is not present
    */
+  @Nullable
   byte[] getGlobalRtDivChunk(int partitionId, byte[] chunkKey);
 
   /**