[controller] Register PARTITION_STATE and STORE_VERSION_STATE schemas at

startup

Add ControllerClientBackedSystemSchemaInitializer for PARTITION_STATE
and STORE_VERSION_STATE, matching the existing pattern for
KAFKA_MESSAGE_ENVELOPE. This allows any child controller to register
new schema versions at startup, removing the dependency on deploying
the system schema cluster controller first.

Gated by new config
controller.storage.protocol.schema.startup.registration.enabled
(default: false) for intentional rollout. Requires the existing
system.schema.initialization.at.start.time.enabled to also be true.

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

Author: pthirun

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

@@ -716,6 +716,15 @@ private ConfigKeys() {
   public static final String SYSTEM_SCHEMA_INITIALIZATION_AT_START_TIME_ENABLED =
       "system.schema.initialization.at.start.time.enabled";
 
+  /**
+   * Whether to register PARTITION_STATE and STORE_VERSION_STATE schemas via
+   * ControllerClientBackedSystemSchemaInitializer at controller startup.
+   * Requires {@link #SYSTEM_SCHEMA_INITIALIZATION_AT_START_TIME_ENABLED} to also be true.
+   * Default: false.
+   */
+  public static final String CONTROLLER_STORAGE_PROTOCOL_SCHEMA_STARTUP_REGISTRATION_ENABLED =
+      "controller.storage.protocol.schema.startup.registration.enabled";
+
   public static final String KME_REGISTRATION_FROM_MESSAGE_HEADER_ENABLED =
       "kme.registration.from.message.header.enabled";
 

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceController.java

@@ -515,6 +515,38 @@ private void initializeSystemSchema(Admin admin) {
               d2ZkHost,
               sslOnly);
       kmeSchemaInitializer.execute();
+
+      if (systemStoreClusterConfig.isStorageProtocolSchemaStartupRegistrationEnabled()) {
+        ControllerClientBackedSystemSchemaInitializer partitionStateSchemaInitializer =
+            new ControllerClientBackedSystemSchemaInitializer(
+                AvroProtocolDefinition.PARTITION_STATE,
+                systemStoreCluster,
+                null,
+                null,
+                false,
+                ((VeniceHelixAdmin) admin).getSslFactory(),
+                childControllerUrl,
+                d2ServiceName,
+                regionD2Client,
+                d2ZkHost,
+                sslOnly);
+        partitionStateSchemaInitializer.execute();
+
+        ControllerClientBackedSystemSchemaInitializer storeVersionStateSchemaInitializer =
+            new ControllerClientBackedSystemSchemaInitializer(
+                AvroProtocolDefinition.STORE_VERSION_STATE,
+                systemStoreCluster,
+                null,
+                null,
+                false,
+                ((VeniceHelixAdmin) admin).getSslFactory(),
+                childControllerUrl,
+                d2ServiceName,
+                regionD2Client,
+                d2ZkHost,
+                sslOnly);
+        storeVersionStateSchemaInitializer.execute();
+      }
     }
   }
 

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceControllerClusterConfig.java

@@ -101,6 +101,7 @@
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_SCHEMA_VALIDATION_ENABLED;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_SSL_ENABLED;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_STORAGE_CLUSTER_HELIX_CLOUD_ENABLED;
+import static com.linkedin.venice.ConfigKeys.CONTROLLER_STORAGE_PROTOCOL_SCHEMA_STARTUP_REGISTRATION_ENABLED;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_STORE_GRAVEYARD_CLEANUP_DELAY_MINUTES;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_STORE_GRAVEYARD_CLEANUP_ENABLED;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_STORE_GRAVEYARD_CLEANUP_SLEEP_INTERVAL_BETWEEN_LIST_FETCH_MINUTES;
@@ -481,6 +482,8 @@ public class VeniceControllerClusterConfig {
 
   private final boolean systemSchemaInitializationAtStartTimeEnabled;
 
+  private final boolean storageProtocolSchemaStartupRegistrationEnabled;
+
   private final boolean isKMERegistrationFromMessageHeaderEnabled;
   private final boolean producerTimestampFallbackEnabled;
   private final boolean unusedValueSchemaCleanupServiceEnabled;
@@ -1194,6 +1197,8 @@ public VeniceControllerClusterConfig(VeniceProperties props) {
         props.getBoolean(CONTROLLER_PARENT_EXTERNAL_SUPERSET_SCHEMA_GENERATION_ENABLED, false);
     this.systemSchemaInitializationAtStartTimeEnabled =
         props.getBoolean(SYSTEM_SCHEMA_INITIALIZATION_AT_START_TIME_ENABLED, false);
+    this.storageProtocolSchemaStartupRegistrationEnabled =
+        props.getBoolean(CONTROLLER_STORAGE_PROTOCOL_SCHEMA_STARTUP_REGISTRATION_ENABLED, false);
     this.isKMERegistrationFromMessageHeaderEnabled =
         props.getBoolean(KME_REGISTRATION_FROM_MESSAGE_HEADER_ENABLED, false);
     this.producerTimestampFallbackEnabled = props.getBoolean(PUBSUB_PRODUCER_TIMESTAMP_FALLBACK_ENABLED, true);
@@ -2206,6 +2211,10 @@ public boolean isSystemSchemaInitializationAtStartTimeEnabled() {
     return systemSchemaInitializationAtStartTimeEnabled;
   }
 
+  public boolean isStorageProtocolSchemaStartupRegistrationEnabled() {
+    return storageProtocolSchemaStartupRegistrationEnabled;
+  }
+
   public boolean isKMERegistrationFromMessageHeaderEnabled() {
     return isKMERegistrationFromMessageHeaderEnabled;
   }