[controller] Checking parent controller region state before handling requests for all routes

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

@@ -1066,6 +1066,17 @@ private ConfigKeys() {
   /** Whether current controller is parent or not */
   public static final String CONTROLLER_PARENT_MODE = "controller.parent.mode";
 
+  /**
+   * This config specifies the state of the region of the parent controller.
+   *
+   * The region can be in one of the following states:
+   * ACTIVE: the parent controller in the region is serving requests.
+   * PASSIVE: the parent controller in the region is rejecting requests.
+   *
+   * By default, this is set to ACTIVE.
+   */
+  public static final String CONTROLLER_PARENT_REGION_STATE = "controller.parent.region.state";
+
   /**
    * This config is used to control how many errored topics we are going to keep in parent cluster.
    * This is mostly used to investigate the Kafka missing message issue.

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/server/TestAdminSparkServer.java

@@ -7,6 +7,7 @@
 import com.linkedin.venice.LastSucceedExecutionIdResponse;
 import com.linkedin.venice.common.VeniceSystemStoreType;
 import com.linkedin.venice.controller.Admin;
+import com.linkedin.venice.controller.ParentControllerRegionState;
 import com.linkedin.venice.controller.VeniceHelixAdmin;
 import com.linkedin.venice.controllerapi.AdminCommandExecution;
 import com.linkedin.venice.controllerapi.ControllerApiConstants;
@@ -98,6 +99,7 @@ public void setUp() {
     extraProperties
         .put(ConfigKeys.CONTROLLER_DEFAULT_MAX_RECORD_SIZE_BYTES, Integer.toString(TEST_MAX_RECORD_SIZE_BYTES));
     extraProperties.put(ConfigKeys.MIN_NUMBER_OF_UNUSED_KAFKA_TOPICS_TO_PRESERVE, Integer.toString(1));
+    extraProperties.put(ConfigKeys.CONTROLLER_PARENT_REGION_STATE, ParentControllerRegionState.ACTIVE.name());
     super.setUp(false, Optional.empty(), extraProperties);
   }
 

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/server/TestAdminSparkWithMocks.java

@@ -6,6 +6,7 @@
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.eq;
 
+import com.linkedin.venice.controller.ParentControllerRegionState;
 import com.linkedin.venice.controller.VeniceHelixAdmin;
 import com.linkedin.venice.controllerapi.ControllerApiConstants;
 import com.linkedin.venice.controllerapi.ControllerRoute;
@@ -205,6 +206,7 @@ public void testAAIncrementalPushRTSourceRegion(boolean sourceGridFabricPresent,
     mockStore.setIncrementalPushEnabled(true);
     doReturn(mockStore).when(admin).getStore(anyString(), anyString());
     doReturn(true).when(admin).isParent();
+    doReturn(ParentControllerRegionState.ACTIVE).when(admin).getParentControllerRegionState();
     doReturn(true).when(admin).isLeaderControllerFor(anyString());
     doReturn(1).when(admin).getReplicationFactor(anyString(), anyString());
     doReturn(1).when(admin).calculateNumberOfPartitions(anyString(), anyString());
@@ -330,6 +332,7 @@ public void testSamzaReplicationPolicyMode(boolean samzaPolicy, boolean storePol
     doReturn("kafka-bootstrap").when(admin).getKafkaBootstrapServers(anyBoolean());
     doReturn("store_rt").when(admin).getRealTimeTopic(anyString(), anyString());
     doReturn(samzaPolicy).when(admin).isParent();
+    doReturn(ParentControllerRegionState.ACTIVE).when(admin).getParentControllerRegionState();
     doReturn(aaEnabled).when(admin).isActiveActiveReplicationEnabledInAllRegion(anyString(), anyString(), eq(true));
     mockStore.setActiveActiveReplicationEnabled(aaEnabled);
 

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/server/TestBackupControllerResponse.java

@@ -4,7 +4,9 @@
 import static com.linkedin.venice.controllerapi.ControllerRoute.JOB;
 import static com.linkedin.venice.controllerapi.ControllerRoute.REQUEST_TOPIC;
 
+import com.linkedin.venice.ConfigKeys;
 import com.linkedin.venice.HttpConstants;
+import com.linkedin.venice.controller.ParentControllerRegionState;
 import com.linkedin.venice.controllerapi.ControllerTransport;
 import com.linkedin.venice.controllerapi.QueryParams;
 import com.linkedin.venice.controllerapi.VersionCreationResponse;
@@ -17,6 +19,7 @@
 import com.linkedin.venice.integration.utils.VeniceControllerWrapper;
 import com.linkedin.venice.integration.utils.ZkServerWrapper;
 import java.util.Optional;
+import java.util.Properties;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeoutException;
 import org.testng.Assert;
@@ -27,6 +30,8 @@ public class TestBackupControllerResponse {
   @Test
   public void backupControllerThrows421() throws Exception {
     String clusterName = "backupControllerThrows421";
+    Properties extraProperties = new Properties();
+    extraProperties.put(ConfigKeys.CONTROLLER_PARENT_REGION_STATE, ParentControllerRegionState.ACTIVE.name());
     try (ZkServerWrapper zkServer = ServiceFactory.getZkServer();
         PubSubBrokerWrapper pubSubBrokerWrapper = ServiceFactory.getPubSubBroker(
             new PubSubBrokerConfigs.Builder().setZkWrapper(zkServer)
@@ -36,10 +41,12 @@ public void backupControllerThrows421() throws Exception {
         VeniceControllerWrapper controller1 = ServiceFactory.getVeniceController(
             new VeniceControllerCreateOptions.Builder(clusterName, zkServer, pubSubBrokerWrapper)
                 .regionName(VeniceClusterWrapperConstants.STANDALONE_REGION_NAME)
+                .extraProperties(extraProperties)
                 .build());
         VeniceControllerWrapper controller2 = ServiceFactory.getVeniceController(
             new VeniceControllerCreateOptions.Builder(clusterName, zkServer, pubSubBrokerWrapper)
                 .regionName(VeniceClusterWrapperConstants.STANDALONE_REGION_NAME)
+                .extraProperties(extraProperties)
                 .build())) {
       // TODO: Eliminate sleep to make test reliable
       Thread.sleep(2000);

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

@@ -755,6 +755,13 @@ void updateRoutersClusterConfig(
    */
   boolean isParent();
 
+  /**
+   * Return the state of the region of the parent controller.
+   * @return {@link ParentControllerRegionState#ACTIVE} which means that the parent controller in the region is serving requests.
+   * Otherwise, return {@link ParentControllerRegionState#PASSIVE}
+   */
+  ParentControllerRegionState getParentControllerRegionState();
+
   /**
    * Get child datacenter to child controller url mapping.
    * @return A map of child datacenter -> child controller url

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

@@ -0,0 +1,16 @@
+package com.linkedin.venice.controller;
+
+/**
+ * Enum representing the state of the region where the parent controller resides.
+ * i.e., Region dc-0 is ACTIVE while Region dc-1 is PASSIVE
+ * This means that ParentController in dc-0 is serving requests while ParentController in dc-1 is rejecting requests
+ */
+public enum ParentControllerRegionState {
+  /** The region is active, so the parent controller in the region is serving requests */
+  ACTIVE,
+  /**
+   * The region is passive, so the parent controller in the region is rejecting requests.
+   * This region is ready to take over if components in the currently active region fails
+   */
+  PASSIVE
+}

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

@@ -53,6 +53,7 @@
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_NAME;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_PARENT_EXTERNAL_SUPERSET_SCHEMA_GENERATION_ENABLED;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_PARENT_MODE;
+import static com.linkedin.venice.ConfigKeys.CONTROLLER_PARENT_REGION_STATE;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_PARENT_SYSTEM_STORE_HEARTBEAT_CHECK_WAIT_TIME_SECONDS;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_PARENT_SYSTEM_STORE_REPAIR_CHECK_INTERVAL_SECONDS;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_PARENT_SYSTEM_STORE_REPAIR_RETRY_COUNT;
@@ -157,6 +158,7 @@
 import static com.linkedin.venice.SSLConfig.DEFAULT_CONTROLLER_SSL_ENABLED;
 import static com.linkedin.venice.VeniceConstants.DEFAULT_PER_ROUTER_READ_QUOTA;
 import static com.linkedin.venice.VeniceConstants.DEFAULT_SSL_FACTORY_CLASS_NAME;
+import static com.linkedin.venice.controller.ParentControllerRegionState.ACTIVE;
 import static com.linkedin.venice.pubsub.PubSubConstants.DEFAULT_KAFKA_REPLICATION_FACTOR;
 import static com.linkedin.venice.pubsub.PubSubConstants.PUBSUB_TOPIC_MANAGER_METADATA_FETCHER_CONSUMER_POOL_SIZE_DEFAULT_VALUE;
 import static com.linkedin.venice.utils.ByteUtils.BYTES_PER_MB;
@@ -221,6 +223,7 @@ public class VeniceControllerClusterConfig {
   private final String controllerClusterZkAddress;
   private final boolean multiRegion;
   private final boolean parent;
+  private final ParentControllerRegionState parentControllerRegionState;
   private final Map<String, String> childDataCenterControllerUrlMap;
   private final String d2ServiceName;
   private final String clusterDiscoveryD2ServiceName;
@@ -631,6 +634,8 @@ public VeniceControllerClusterConfig(VeniceProperties props) {
     this.controllerClusterReplica = props.getInt(CONTROLLER_CLUSTER_REPLICA, 3);
     this.controllerClusterZkAddress = props.getString(CONTROLLER_CLUSTER_ZK_ADDRESSS, getZkAddress());
     this.parent = props.getBoolean(CONTROLLER_PARENT_MODE, false);
+    this.parentControllerRegionState =
+        ParentControllerRegionState.valueOf(props.getString(CONTROLLER_PARENT_REGION_STATE, ACTIVE.name()));
 
     if (childDatacenters.isEmpty()) {
       this.childDataCenterControllerUrlMap = Collections.emptyMap();
@@ -1162,6 +1167,10 @@ public boolean isParent() {
     return parent;
   }
 
+  public ParentControllerRegionState getParentControllerRegionState() {
+    return parentControllerRegionState;
+  }
+
   public long getDeprecatedJobTopicRetentionMs() {
     return deprecatedJobTopicRetentionMs;
   }

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

@@ -67,6 +67,10 @@ public boolean isParent() {
     return getCommonConfig().isParent();
   }
 
+  public ParentControllerRegionState getParentControllerRegionState() {
+    return getCommonConfig().getParentControllerRegionState();
+  }
+
   public String getControllerName() {
     return getCommonConfig().getControllerName();
   }

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

@@ -7553,6 +7553,14 @@ public boolean isParent() {
     return multiClusterConfigs.isParent();
   }
 
+  /**
+   * @see Admin#getParentControllerRegionState()
+   */
+  @Override
+  public ParentControllerRegionState getParentControllerRegionState() {
+    return multiClusterConfigs.getParentControllerRegionState();
+  }
+
   /**
    * @see Admin#getChildDataCenterControllerUrlMap(String)
    */

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

@@ -4871,6 +4871,14 @@ public boolean isParent() {
     return getVeniceHelixAdmin().isParent();
   }
 
+  /**
+   * @see Admin#getParentControllerRegionState()
+   */
+  @Override
+  public ParentControllerRegionState getParentControllerRegionState() {
+    return getVeniceHelixAdmin().getParentControllerRegionState();
+  }
+
   /**
    * @see Admin#getChildDataCenterControllerUrlMap(String)
    */