[server][common] Server read quota initialization without CV fallback strategy (#2352)

Problem Statement:
1. If one store version is not initialized correctly and throws unhandled exception, we would skip the initialization of follow up stores in the storeRepository
2. If CV repository is delayed or unavailable (which we have been seeing more frequently recently especially with large clusters) then we often fail quota initialization due to VeniceNoHelixResourceException which was also unhandled as per problem 1.
3. Having both initialized and initializedVolatile is confusing and unnecessary.

Solution:
1. Catch unhandled exceptions at the store level and continue initialization for other stores to isolate any initialization errors.
2. If CV is unavailable, i.e. VeniceNoHelixResourceException is thrown or partition assignment map is somehow empty and fallback strategy is enabled (enabled by default but can be disabled via config server.read.quota.initialization.fallback.enabled). We will allocate this instance Q * X/P quota where Q is the total read quota, X is the number of partitions assigned to the node based on storage engine state and P is the store version partition count.
3. Removed initialized flag and only keeping the initializedVolatile flag.
4. Added additional logging to capture only when there is a total quota change and how is the new rate limiter calculated. i.e. the total quota, node responsibility and resulting instance quota. Also added additional logging whenever fallback strategy is invoked regardless of the quota update trigger (from store update or CV update events).

Author: xunyin8

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

@@ -175,6 +175,7 @@
 import static com.linkedin.venice.ConfigKeys.SERVER_QUOTA_ENFORCEMENT_CAPACITY_MULTIPLE;
 import static com.linkedin.venice.ConfigKeys.SERVER_QUOTA_ENFORCEMENT_ENABLED;
 import static com.linkedin.venice.ConfigKeys.SERVER_QUOTA_ENFORCEMENT_INTERVAL_IN_MILLIS;
+import static com.linkedin.venice.ConfigKeys.SERVER_READ_QUOTA_INITIALIZATION_FALLBACK_ENABLED;
 import static com.linkedin.venice.ConfigKeys.SERVER_RECORD_LEVEL_METRICS_WHEN_BOOTSTRAPPING_CURRENT_VERSION_ENABLED;
 import static com.linkedin.venice.ConfigKeys.SERVER_REMOTE_CONSUMER_CONFIG_PREFIX;
 import static com.linkedin.venice.ConfigKeys.SERVER_REMOTE_INGESTION_REPAIR_SLEEP_INTERVAL_SECONDS;
@@ -691,6 +692,7 @@ public class VeniceServerConfig extends VeniceClusterConfig {
 
   private final boolean parallelResourceShutdownEnabled;
   private final int lagMonitorCleanupCycle;
+  private final boolean readQuotaInitializationFallbackEnabled;
 
   public VeniceServerConfig(VeniceProperties serverProperties) throws ConfigurationException {
     this(serverProperties, Collections.emptyMap());
@@ -1177,6 +1179,8 @@ public VeniceServerConfig(VeniceProperties serverProperties, Map<String, Map<Str
         serverProperties.getBoolean(SERVER_PARALLEL_RESOURCE_SHUTDOWN_ENABLED, false);
     this.lagMonitorCleanupCycle =
         serverProperties.getInt(SERVER_LAG_MONITOR_CLEANUP_CYCLE, DEFAULT_LAG_MONITOR_CLEANUP_CYCLE);
+    this.readQuotaInitializationFallbackEnabled =
+        serverProperties.getBoolean(SERVER_READ_QUOTA_INITIALIZATION_FALLBACK_ENABLED, true);
   }
 
   List<Double> extractThrottleLimitFactorsFor(VeniceProperties serverProperties, String configKey) {
@@ -2131,4 +2135,8 @@ public boolean isParallelResourceShutdownEnabled() {
   public int getLagMonitorCleanupCycle() {
     return lagMonitorCleanupCycle;
   }
+
+  public boolean isReadQuotaInitializationFallbackEnabled() {
+    return readQuotaInitializationFallbackEnabled;
+  }
 }

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

@@ -3085,4 +3085,11 @@ private ConfigKeys() {
    */
   public static final String SYSTEM_STORE_VERSION_RETENTION_COUNT = "store.version.retention.count.system.store";
   public static final int DEFAULT_SYSTEM_STORE_VERSION_RETENTION_COUNT = 5;
+
+  /**
+   * Whether storage node read quota will fail-open if CV is unavailable during initialization or it will try to use
+   * cluster's instance count to calculate a placeholder/fallback value until CV is available.
+   */
+  public static final String SERVER_READ_QUOTA_INITIALIZATION_FALLBACK_ENABLED =
+      "server.read.quota.initialization.fallback.enabled";
 }

services/venice-server/src/main/java/com/linkedin/venice/listener/HttpChannelInitializer.java

@@ -4,6 +4,7 @@
 import com.linkedin.alpini.netty4.http2.Http2PipelineInitializer;
 import com.linkedin.alpini.netty4.ssl.SslInitializer;
 import com.linkedin.davinci.config.VeniceServerConfig;
+import com.linkedin.davinci.storage.StorageEngineRepository;
 import com.linkedin.venice.acl.DynamicAccessController;
 import com.linkedin.venice.acl.StaticAccessController;
 import com.linkedin.venice.authorization.IdentityParser;
@@ -77,7 +78,8 @@ public HttpChannelInitializer(
       VeniceServerConfig serverConfig,
       Optional<StaticAccessController> routerAccessController,
       Optional<DynamicAccessController> storeAccessController,
-      StorageReadRequestHandler requestHandler) {
+      StorageReadRequestHandler requestHandler,
+      StorageEngineRepository storageEngineRepository) {
     this.serverConfig = serverConfig;
     this.requestHandler = requestHandler;
     this.isDaVinciClient = serverConfig.isDaVinciClient();
@@ -146,6 +148,7 @@ public HttpChannelInitializer(
           serverConfig,
           storeMetadataRepository,
           customizedViewRepository,
+          storageEngineRepository,
           nodeId,
           quotaUsageStats);
     } else {

services/venice-server/src/main/java/com/linkedin/venice/listener/ListenerService.java

@@ -125,7 +125,8 @@ public ListenerService(
         serverConfig,
         routerAccessController,
         storeAccessController,
-        requestHandler);
+        requestHandler,
+        storageEngineRepository);
 
     Class<? extends ServerChannel> serverSocketChannelClass = NioServerSocketChannel.class;
     boolean epollEnabled = serverConfig.isRestServiceEpollEnabled();