Add feature flag

Author: Minh Nguyen

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

@@ -530,6 +530,13 @@ private ConfigKeys() {
   public static final String CHILD_CONTROLLER_ADMIN_TOPIC_CONSUMPTION_ENABLED =
       "child.controller.admin.topic.consumption.enabled";
 
+  /**
+   * Whether to enable admin operation system store or not.
+   * If yes, controller will register admin operation system store and process admin operations in rollback cases.
+   */
+  public static final String CONTROLLER_ADMIN_OPERATION_SYSTEM_STORE_ENABLED =
+      "controller.admin.operation.system.store.enabled";
+
   /**
    * This config defines the source region of aggregate hybrid store real-time data when native replication is enabled
    */

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/kafka/consumer/AdminConsumptionWithProtocolRollbackIntegrationTest.java

@@ -1,7 +1,6 @@
 package com.linkedin.venice.controller.kafka.consumer;
 
-import static com.linkedin.venice.ConfigKeys.ADMIN_CONSUMPTION_CYCLE_TIMEOUT_MS;
-import static com.linkedin.venice.ConfigKeys.ADMIN_CONSUMPTION_MAX_WORKER_THREAD_POOL_SIZE;
+import static com.linkedin.venice.ConfigKeys.*;
 
 import com.linkedin.venice.controller.VeniceParentHelixAdmin;
 import com.linkedin.venice.controller.kafka.AdminTopicUtils;
@@ -83,6 +82,10 @@ public void setUp() {
     Properties parentControllerProps = new Properties();
     parentControllerProps.put(ADMIN_CONSUMPTION_MAX_WORKER_THREAD_POOL_SIZE, adminConsumptionMaxWorkerPoolSize);
     parentControllerProps.put(ADMIN_CONSUMPTION_CYCLE_TIMEOUT_MS, 3000);
+    parentControllerProps.put(CONTROLLER_ADMIN_OPERATION_SYSTEM_STORE_ENABLED, true);
+
+    Properties childControllerProps = new Properties();
+    childControllerProps.put(CONTROLLER_ADMIN_OPERATION_SYSTEM_STORE_ENABLED, true);
     venice = ServiceFactory.getVeniceTwoLayerMultiRegionMultiClusterWrapper(
         new VeniceMultiRegionClusterCreateOptions.Builder().numberOfRegions(1)
             .numberOfClusters(1)
@@ -92,6 +95,7 @@ public void setUp() {
             .numberOfRouters(1)
             .replicationFactor(1)
             .parentControllerProperties(parentControllerProps)
+            .childControllerProperties(childControllerProps)
             .serverProperties(serverProperties)
             .build());
 

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

@@ -659,6 +659,11 @@ public class VeniceControllerClusterConfig {
   private final boolean useMultiRegionRealTimeTopicSwitcher;
   private final Set<String> activeActiveRealTimeSourceFabrics;
 
+  /**
+   * Admin operation system store
+   */
+  private final boolean isAdminOperationSystemStoreEnabled;
+
   public VeniceControllerClusterConfig(VeniceProperties props) {
     this.props = props;
     this.clusterName = props.getString(CLUSTER_NAME);
@@ -1241,6 +1246,8 @@ public VeniceControllerClusterConfig(VeniceProperties props) {
         props.getBoolean(CONTROLLER_BACKUP_VERSION_REPLICA_REDUCTION_ENABLED, false);
     this.useMultiRegionRealTimeTopicSwitcher =
         props.getBoolean(ConfigKeys.CONTROLLER_USE_MULTI_REGION_REAL_TIME_TOPIC_SWITCHER_ENABLED, false);
+    this.isAdminOperationSystemStoreEnabled =
+        props.getBoolean(ConfigKeys.CONTROLLER_ADMIN_OPERATION_SYSTEM_STORE_ENABLED, false);
 
     this.logClusterConfig();
   }
@@ -2185,6 +2192,10 @@ public boolean isRealTimeTopicVersioningEnabled() {
     return isRealTimeTopicVersioningEnabled;
   }
 
+  public boolean isAdminOperationSystemStoreEnabled() {
+    return isAdminOperationSystemStoreEnabled;
+  }
+
   /**
    * A function that would put a k/v pair into a map with some processing works.
    */

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

@@ -275,6 +275,10 @@ public boolean isZkSharedMetaSystemSchemaStoreAutoCreationEnabled() {
     return getCommonConfig().isZkSharedMetaSystemSchemaStoreAutoCreationEnabled();
   }
 
+  public boolean isAdminOperationSystemStoreEnabled() {
+    return getCommonConfig().isAdminOperationSystemStoreEnabled();
+  }
+
   public boolean isZkSharedDaVinciPushStatusSystemSchemaStoreAutoCreationEnabled() {
     return getCommonConfig().isZkSharedDaVinciPushStatusSystemSchemaStoreAutoCreationEnabled();
   }

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

@@ -697,15 +697,17 @@ public VeniceHelixAdmin(
             AvroProtocolDefinition.SERVER_STORE_PROPERTIES_PAYLOAD,
             multiClusterConfigs,
             this));
-    initRoutines.add(
-        new SystemSchemaInitializationRoutine(
-            AvroProtocolDefinition.ADMIN_OPERATION,
-            multiClusterConfigs,
-            this,
-            Optional.empty(),
-            Optional.empty(),
-            false,
-            AdminOperation.class));
+    if (multiClusterConfigs.isAdminOperationSystemStoreEnabled()) {
+      initRoutines.add(
+          new SystemSchemaInitializationRoutine(
+              AvroProtocolDefinition.ADMIN_OPERATION,
+              multiClusterConfigs,
+              this,
+              Optional.empty(),
+              Optional.empty(),
+              false,
+              AdminOperation.class));
+    }
 
     if (multiClusterConfigs.isZkSharedMetaSystemSchemaStoreAutoCreationEnabled()) {
       // Add routine to create zk shared metadata system store
@@ -1116,6 +1118,10 @@ protected HelixAdmin getHelixAdmin() {
     return this.admin;
   }
 
+  public boolean isAdminOperationSystemStoreEnabled() {
+    return multiClusterConfigs.isAdminOperationSystemStoreEnabled();
+  }
+
   /**
    * Create a new ZK store and its configuration in the store repository and create schemas in the schema repository.
    * @param clusterName Venice cluster where the store locates.

services/venice-controller/src/main/java/com/linkedin/venice/controller/kafka/consumer/AdminConsumptionTask.java

@@ -902,7 +902,9 @@ private Pair<Long, AdminOperation> extractExecutionIdAndAndAdminOperation(Defaul
     if (MessageType.PUT == messageType) {
       Put put = (Put) kafkaValue.payloadUnion;
       try {
-        deserializer.fetchAndStoreSchemaIfAbsent(admin, put.schemaId);
+        if (admin.isAdminOperationSystemStoreEnabled()) {
+          deserializer.fetchAndStoreSchemaIfAbsent(admin, put.schemaId);
+        }
         adminOperation = deserializer.deserialize(put.putValue, put.schemaId);
         executionIdFromPayload = adminOperation.executionId;
       } catch (Exception e) {