Fix regression: use legacy rebalancer in

addStorageClusterResourceToControllerCluster

The previous fix called addVeniceStorageClusterToControllerCluster() from
createClusterIfRequired(), which uses WagedRebalancer. WagedRebalancer
requires instance capacity configuration not present in integration tests,
so waitUntilClusterResourceIsVisibleInEV() blocked for 5 minutes, causing
all integration tests that start a Venice cluster to hang and hit their
TestNG timeouts.

Fix: introduce addStorageClusterResourceToControllerCluster(clusterName,
replicaCount)
in HelixAdminClient/ZkHelixAdminClient. This method:
- Uses controllerClusterHelixAdmin (routes to controller.cluster.zk.address)
  so it is correct in both shared-ZK and split-ZK deployments
- Keeps the original DelayedAutoRebalancer + CrushRebalanceStrategy config
  from the old createClusterIfRequired implementation, preserving assignment
  behavior for the non-HAAS path

createClusterIfRequired now calls
helixAdminClient.addStorageClusterResourceToControllerCluster
instead of the direct admin.addResource/setResourceIdealState/rebalance calls.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: nvijayak

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

@@ -51,6 +51,22 @@ public interface HelixAdminClient {
    */
   void addVeniceStorageClusterToControllerCluster(String clusterName);
 
+  /**
+   * Register the given Venice storage cluster as a resource in the controller cluster using the
+   * non-HAAS rebalancer configuration ({@code DelayedAutoRebalancer} +
+   * {@code CrushRebalanceStrategy}). Must route to the controller-cluster ZK (not the storage ZK)
+   * so that the operation is visible to the Helix manager connected via
+   * {@code controller.cluster.zk.address} in split-ZK deployments.
+   *
+   * <p>Used by {@link VeniceHelixAdmin#createClusterIfRequired} which handles the non-HAAS storage
+   * cluster path. Unlike {@link #addVeniceStorageClusterToControllerCluster} (which uses
+   * {@code WagedRebalancer}), this method preserves the original rebalancer used in the legacy path.
+   *
+   * @param clusterName  name of the Venice storage cluster being registered
+   * @param replicaCount number of replicas (controller instances) for this resource
+   */
+  void addStorageClusterResourceToControllerCluster(String clusterName, int replicaCount);
+
   /**
    * Check if the grand cluster managed by HaaS controllers is aware of the given cluster.
    * @param clusterName of the cluster.

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

@@ -7292,12 +7292,10 @@ private void createClusterIfRequired(String clusterName) {
         delayedTime);
     admin.addStateModelDef(clusterName, LeaderStandbySMD.name, LeaderStandbySMD.build());
 
-    // Register the storage cluster as a resource in the controller cluster. Must use
-    // helixAdminClient (routes to controller.cluster.zk.address) rather than admin (storage ZK),
-    // because in split-ZK deployments the controller cluster lives on the controller ZK.
-    if (!helixAdminClient.isVeniceStorageClusterInControllerCluster(clusterName)) {
-      helixAdminClient.addVeniceStorageClusterToControllerCluster(clusterName);
-    }
+    // Register the storage cluster as a resource in the controller cluster using the legacy
+    // (non-HAAS) rebalancer config. Must route to the controller ZK in split-ZK deployments.
+    int controllerClusterReplica = config.getControllerClusterReplica();
+    helixAdminClient.addStorageClusterResourceToControllerCluster(clusterName, controllerClusterReplica);
   }
 
   public boolean updateIdealState(String clusterName, String resourceName, int minReplica) {

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

@@ -17,6 +17,7 @@
 import org.apache.helix.constants.InstanceConstants;
 import org.apache.helix.controller.rebalancer.DelayedAutoRebalancer;
 import org.apache.helix.controller.rebalancer.strategy.AutoRebalanceStrategy;
+import org.apache.helix.controller.rebalancer.strategy.CrushRebalanceStrategy;
 import org.apache.helix.controller.rebalancer.waged.WagedRebalancer;
 import org.apache.helix.manager.zk.ZKHelixAdmin;
 import org.apache.helix.manager.zk.ZKHelixManager;
@@ -291,8 +292,32 @@ public void addVeniceStorageClusterToControllerCluster(String clusterName) {
   }
 
   /**
-   * @see HelixAdminClient#isClusterInGrandCluster(String)
+   * @see HelixAdminClient#addStorageClusterResourceToControllerCluster(String, int)
    */
+  @Override
+  public void addStorageClusterResourceToControllerCluster(String clusterName, int replicaCount) {
+    if (isVeniceStorageClusterInControllerCluster(clusterName)) {
+      return;
+    }
+    // Use controllerClusterHelixAdmin (routes to controller.cluster.zk.address) so this works in
+    // both shared-ZK and split-ZK deployments. Keep the legacy non-HAAS rebalancer config so
+    // that the assignment algorithm is unchanged for this path.
+    controllerClusterHelixAdmin.addResource(
+        controllerClusterName,
+        clusterName,
+        CONTROLLER_CLUSTER_PARTITION_COUNT,
+        LeaderStandbySMD.name,
+        IdealState.RebalanceMode.FULL_AUTO.toString(),
+        AutoRebalanceStrategy.class.getName());
+    IdealState idealState = controllerClusterHelixAdmin.getResourceIdealState(controllerClusterName, clusterName);
+    idealState.setReplicas(String.valueOf(replicaCount));
+    idealState.setMinActiveReplicas(replicaCount);
+    idealState.setRebalancerClassName(DelayedAutoRebalancer.class.getName());
+    idealState.setRebalanceStrategy(CrushRebalanceStrategy.class.getName());
+    controllerClusterHelixAdmin.setResourceIdealState(controllerClusterName, clusterName, idealState);
+    controllerClusterHelixAdmin.rebalance(controllerClusterName, clusterName, replicaCount);
+  }
+
   @Override
   public boolean isClusterInGrandCluster(String clusterName) {
     return controllerClusterHelixAdmin.getResourcesInCluster(haasSuperClusterName).contains(clusterName);