[da-vinci][server] Move transient record cache flag to version level and add

host-level kill switch

- Move transientRecordCacheEnabled from Store interface to Version interface
so
  the cache can be toggled per version without affecting other versions.
- Add SERVER_INGESTION_TRANSIENT_RECORD_CACHE_ENABLED server config (default
true)
  as a host-level kill switch to disable the cache on a node without a new
push.
- Fix hot cache staleness: invalidate hot cache entry when a key is
overwritten
  with a value below the size threshold or a null-value update.
- Add tests for cache invalidation on small-value and null-value overwrites.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: sixpluszero

clients/da-vinci-client/src/main/java/com/linkedin/davinci/config/VeniceServerConfig.java

@@ -132,6 +132,7 @@
 import static com.linkedin.venice.ConfigKeys.SERVER_INGESTION_OTEL_STATS_ENABLED;
 import static com.linkedin.venice.ConfigKeys.SERVER_INGESTION_TASK_MAX_IDLE_COUNT;
 import static com.linkedin.venice.ConfigKeys.SERVER_INGESTION_TASK_REUSABLE_OBJECTS_STRATEGY;
+import static com.linkedin.venice.ConfigKeys.SERVER_INGESTION_TRANSIENT_RECORD_CACHE_ENABLED;
 import static com.linkedin.venice.ConfigKeys.SERVER_INGESTION_TRANSIENT_RECORD_CACHE_MAX_WEIGHT;
 import static com.linkedin.venice.ConfigKeys.SERVER_INGESTION_TRANSIENT_RECORD_CACHE_MIN_VALUE_SIZE;
 import static com.linkedin.venice.ConfigKeys.SERVER_KAFKA_CONSUMER_OFFSET_COLLECTION_ENABLED;
@@ -713,6 +714,7 @@ public class VeniceServerConfig extends VeniceClusterConfig {
   private final int lagMonitorCleanupCycle;
   private final boolean readQuotaInitializationFallbackEnabled;
   private final boolean ingestionProgressLoggingEnabled;
+  private final boolean transientRecordCacheEnabled;
   private final long transientRecordCacheMaxWeight;
   private final int transientRecordCacheMinValueSize;
 
@@ -1231,6 +1233,8 @@ public VeniceServerConfig(VeniceProperties serverProperties, Map<String, Map<Str
     this.readQuotaInitializationFallbackEnabled =
         serverProperties.getBoolean(SERVER_READ_QUOTA_INITIALIZATION_FALLBACK_ENABLED, true);
     this.ingestionProgressLoggingEnabled = serverProperties.getBoolean(POSITIONAL_PROGRESS_LOGGING_ENABLED, false);
+    this.transientRecordCacheEnabled =
+        serverProperties.getBoolean(SERVER_INGESTION_TRANSIENT_RECORD_CACHE_ENABLED, true);
     this.transientRecordCacheMaxWeight =
         serverProperties.getLong(SERVER_INGESTION_TRANSIENT_RECORD_CACHE_MAX_WEIGHT, 32 * 1024 * 1024L);
     this.transientRecordCacheMinValueSize =
@@ -2234,6 +2238,10 @@ public boolean isIngestionProgressLoggingEnabled() {
     return ingestionProgressLoggingEnabled;
   }
 
+  public boolean isTransientRecordCacheEnabled() {
+    return transientRecordCacheEnabled;
+  }
+
   public long getTransientRecordCacheMaxWeight() {
     return transientRecordCacheMaxWeight;
   }

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

@@ -789,8 +789,14 @@ public void setTransientRecord(
 
     transientRecordMap.put(ByteArrayKey.wrap(key), transientRecord);
 
-    if (hotRecordCache != null && valueLen >= minValueSizeForHotCache) {
-      hotRecordCache.put(buildHotCacheKey(key), transientRecord);
+    if (hotRecordCache != null) {
+      ByteArrayKey hotKey = buildHotCacheKey(key);
+      if (valueLen >= minValueSizeForHotCache) {
+        hotRecordCache.put(hotKey, transientRecord);
+      } else {
+        // Invalidate any stale entry when the new record no longer qualifies
+        hotRecordCache.invalidate(hotKey);
+      }
     }
   }
 

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

@@ -689,7 +689,7 @@ public StoreIngestionTask(
     this.localKafkaServerSingletonSet = Collections.singleton(localKafkaServer);
     this.isActiveActiveReplicationEnabled = version.isActiveActiveReplicationEnabled();
 
-    if (store.isTransientRecordCacheEnabled()
+    if (serverConfig.isTransientRecordCacheEnabled() && version.isTransientRecordCacheEnabled()
         && (this.isActiveActiveReplicationEnabled || this.isWriteComputationEnabled)) {
       long maxWeight = serverConfig.getTransientRecordCacheMaxWeight();
       this.hotRecordCache = Caffeine.newBuilder()

clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/PartitionConsumptionStateTest.java

@@ -348,6 +348,79 @@ public void testHotRecordCacheAdmissionThreshold() {
     assertEquals(pcs.getHotRecordCacheHitCount(), 0);
   }
 
+  @Test
+  public void testHotCacheInvalidatedWhenReplacedWithSmallValue() {
+    Cache<ByteArrayKey, PartitionConsumptionState.TransientRecord> hotCache = Caffeine.newBuilder()
+        .maximumWeight(1024 * 1024)
+        .weigher((ByteArrayKey k, PartitionConsumptionState.TransientRecord v) -> {
+          return k.getContent().length + (v.getValue() != null ? v.getValueLen() : 0);
+        })
+        .build();
+
+    int minValueSize = 50;
+    PartitionConsumptionState pcs = new PartitionConsumptionState(
+        TOPIC_PARTITION,
+        mock(OffsetRecord.class),
+        pubSubContext,
+        false,
+        Schema.create(Schema.Type.STRING),
+        hotCache,
+        minValueSize);
+
+    PubSubPosition pos1 = mock(PubSubPosition.class);
+    PubSubPosition pos2 = mock(PubSubPosition.class);
+    byte[] key = new byte[] { 1, 2, 3 };
+    byte[] largeValue = new byte[100]; // >= minValueSize, admitted to hot cache
+    byte[] smallValue = new byte[10]; // < minValueSize
+
+    // First write: large value gets into hot cache
+    pcs.setTransientRecord(-1, pos1, key, largeValue, 0, largeValue.length, 1, null);
+    pcs.mayRemoveTransientRecord(-1, pos1, key);
+    // Verify hot cache serves it
+    assertNotNull(pcs.getTransientRecord(key));
+
+    // Second write: small value replaces it — hot cache entry must be invalidated
+    pcs.setTransientRecord(-1, pos2, key, smallValue, 0, smallValue.length, 2, null);
+    pcs.mayRemoveTransientRecord(-1, pos2, key);
+    // Stale large-value entry should NOT be returned
+    assertNull(pcs.getTransientRecord(key));
+  }
+
+  @Test
+  public void testHotCacheInvalidatedWhenReplacedWithNullValue() {
+    Cache<ByteArrayKey, PartitionConsumptionState.TransientRecord> hotCache = Caffeine.newBuilder()
+        .maximumWeight(1024 * 1024)
+        .weigher((ByteArrayKey k, PartitionConsumptionState.TransientRecord v) -> {
+          return k.getContent().length + (v.getValue() != null ? v.getValueLen() : 0);
+        })
+        .build();
+
+    int minValueSize = 50;
+    PartitionConsumptionState pcs = new PartitionConsumptionState(
+        TOPIC_PARTITION,
+        mock(OffsetRecord.class),
+        pubSubContext,
+        false,
+        Schema.create(Schema.Type.STRING),
+        hotCache,
+        minValueSize);
+
+    PubSubPosition pos1 = mock(PubSubPosition.class);
+    PubSubPosition pos2 = mock(PubSubPosition.class);
+    byte[] key = new byte[] { 1, 2, 3 };
+    byte[] largeValue = new byte[100];
+
+    // First write: large value gets into hot cache
+    pcs.setTransientRecord(-1, pos1, key, largeValue, 0, largeValue.length, 1, null);
+    pcs.mayRemoveTransientRecord(-1, pos1, key);
+    assertNotNull(pcs.getTransientRecord(key));
+
+    // Second write: null-value overload (valueLen = -1) — must invalidate hot cache entry
+    pcs.setTransientRecord(-1, pos2, key, 2, null);
+    pcs.mayRemoveTransientRecord(-1, pos2, key);
+    assertNull(pcs.getTransientRecord(key));
+  }
+
   @Test
   public void testNullHotCacheBehavesIdentically() {
     // Without hot cache (null), behavior should be identical to previous implementation

internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java

@@ -3351,4 +3351,12 @@ private ConfigKeys() {
    */
   public static final String SERVER_INGESTION_TRANSIENT_RECORD_CACHE_MIN_VALUE_SIZE =
       "server.ingestion.transient.record.cache.min.value.size";
+
+  /**
+   * Host-level kill switch for the transient record cache. When set to false, the cache is disabled
+   * on this node regardless of the version-level setting. This allows operators to quickly disable
+   * the cache without requiring a new push. Default is true (enabled, defers to version-level config).
+   */
+  public static final String SERVER_INGESTION_TRANSIENT_RECORD_CACHE_ENABLED =
+      "server.ingestion.transient.record.cache.enabled";
 }

internal/venice-common/src/main/java/com/linkedin/venice/meta/ReadOnlyStore.java

@@ -1067,7 +1067,6 @@ public StoreProperties cloneStoreProperties() {
     storeProperties.setKeyUrnCompressionEnabled(isKeyUrnCompressionEnabled());
     storeProperties.setKeyUrnFields(getKeyUrnFields().stream().map(String::toString).collect(Collectors.toList()));
     storeProperties.setPreviousCurrentVersion(getPreviousCurrentVersion());
-    storeProperties.setTransientRecordCacheEnabled(isTransientRecordCacheEnabled());
     // Set blobDbEnabled to default value - field exists in schema but not yet exposed via Store interface
     storeProperties.setBlobDbEnabled("NOT_SPECIFIED");
 
@@ -1810,16 +1809,6 @@ public void setPreviousCurrentVersion(int previousCurrentVersion) {
     throw new UnsupportedOperationException();
   }
 
-  @Override
-  public boolean isTransientRecordCacheEnabled() {
-    return this.delegate.isTransientRecordCacheEnabled();
-  }
-
-  @Override
-  public void setTransientRecordCacheEnabled(boolean transientRecordCacheEnabled) {
-    throw new UnsupportedOperationException();
-  }
-
   @Override
   public String toString() {
     return this.delegate.toString();

internal/venice-common/src/main/java/com/linkedin/venice/meta/Store.java

@@ -393,8 +393,4 @@ default IntSet getVersionNumbers() {
   int getPreviousCurrentVersion();
 
   void setPreviousCurrentVersion(int previousCurrentVersion);
-
-  boolean isTransientRecordCacheEnabled();
-
-  void setTransientRecordCacheEnabled(boolean transientRecordCacheEnabled);
 }

internal/venice-common/src/main/java/com/linkedin/venice/meta/StoreInfo.java

@@ -72,7 +72,6 @@ public static StoreInfo fromStore(Store store) {
     storeInfo.setReplicationMetadataVersionId(store.getRmdVersion());
     storeInfo.setViewConfigs(store.getViewConfigs());
     storeInfo.setStorageNodeReadQuotaEnabled(store.isStorageNodeReadQuotaEnabled());
-    storeInfo.setTransientRecordCacheEnabled(store.isTransientRecordCacheEnabled());
     storeInfo.setCompactionEnabled(store.isCompactionEnabled());
     storeInfo.setCompactionThreshold(store.getCompactionThresholdMilliseconds());
     storeInfo.setMinCompactionLagSeconds(store.getMinCompactionLagSeconds());
@@ -339,11 +338,6 @@ public static StoreInfo fromStore(Store store) {
    */
   private boolean storageNodeReadQuotaEnabled;
 
-  /**
-   * Whether the bounded hot transient record cache is enabled for this store.
-   */
-  private boolean transientRecordCacheEnabled = false;
-
   /**
    * Reasons for why or why not the store is dead
    */
@@ -840,14 +834,6 @@ public void setStorageNodeReadQuotaEnabled(boolean storageNodeReadQuotaEnabled)
     this.storageNodeReadQuotaEnabled = storageNodeReadQuotaEnabled;
   }
 
-  public boolean isTransientRecordCacheEnabled() {
-    return transientRecordCacheEnabled;
-  }
-
-  public void setTransientRecordCacheEnabled(boolean transientRecordCacheEnabled) {
-    this.transientRecordCacheEnabled = transientRecordCacheEnabled;
-  }
-
   public boolean isCompactionEnabled() {
     return this.compactionEnabled;
   }

internal/venice-common/src/main/java/com/linkedin/venice/meta/SystemStore.java

@@ -857,16 +857,6 @@ public void setPreviousCurrentVersion(int previousCurrentVersion) {
     throwUnsupportedOperationException("setPreviousCurrentVersion");
   }
 
-  @Override
-  public boolean isTransientRecordCacheEnabled() {
-    return zkSharedStore.isTransientRecordCacheEnabled();
-  }
-
-  @Override
-  public void setTransientRecordCacheEnabled(boolean transientRecordCacheEnabled) {
-    throwUnsupportedOperationException("setTransientRecordCacheEnabled");
-  }
-
   @Override
   public boolean isGlobalRtDivEnabled() {
     return zkSharedStore.isGlobalRtDivEnabled();

internal/venice-common/src/main/java/com/linkedin/venice/meta/Version.java

@@ -329,6 +329,10 @@ default void setTimestampMetadataVersionId(int replicationMetadataVersionId) {
 
   void setPreviousCurrentVersion(int previousCurrentVersion);
 
+  boolean isTransientRecordCacheEnabled();
+
+  void setTransientRecordCacheEnabled(boolean transientRecordCacheEnabled);
+
   /**
    * Kafka topic name is composed by store name and version.
    * <p>