Fix createClusterIfRequired to use controller ZK for controller-cluster ops

In split-ZK mode, the controller cluster (venice-controllers) lives on
controller.cluster.zk.address, not on zookeeper.address. The four
admin.addResource/getResourceIdealState/setResourceIdealState/rebalance
calls in createClusterIfRequired all used admin (storage ZK), causing
'cluster venice-controllers is not setup yet' because that cluster does
not exist on the storage ZK.

Fix: replace those calls with
helixAdminClient.addVeniceStorageClusterToControllerCluster()
which correctly routes to the controller ZK via controllerClusterHelixAdmin.
This is safe in both shared-ZK (no behavioral change, same ZK) and split-ZK
deployments. Also remove now-unused DelayedAutoRebalancer,
CrushRebalanceStrategy
imports and the CONTROLLER_CLUSTER_NUMBER_OF_PARTITION constant.

Also fix TestVeniceControllerZkClientIsolation: change
@BeforeClass/@afterclass
for storageZk and controllerZk to @BeforeMethod/@AfterMethod. The split-ZK
test creates venice-controllers on controllerZk; without per-method ZK
lifecycle, that data leaks into testSharedZkBaseline which then fails its
assertion that controllerZk holds no Venice clusters.

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

Author: nvijayak

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

@@ -34,9 +34,8 @@
 import java.util.Properties;
 import java.util.concurrent.TimeUnit;
 import org.apache.helix.manager.zk.ZKHelixAdmin;
-import org.testng.annotations.AfterClass;
 import org.testng.annotations.AfterMethod;
-import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
 
@@ -74,7 +73,7 @@ public class TestVeniceControllerZkClientIsolation {
   private String clusterName;
   private String controllerClusterName;
 
-  @BeforeClass
+  @BeforeMethod(alwaysRun = true)
   public void setUp() {
     Utils.thisIsLocalhost();
     storageZk = ServiceFactory.getZkServer();
@@ -86,10 +85,6 @@ public void setUp() {
   public void tearDownAdmin() {
     Utils.closeQuietlyWithErrorLogged(veniceAdmin);
     veniceAdmin = null;
-  }
-
-  @AfterClass(alwaysRun = true)
-  public void tearDown() {
     Utils.closeQuietlyWithErrorLogged(pubSubBrokerWrapper);
     Utils.closeQuietlyWithErrorLogged(controllerZk);
     Utils.closeQuietlyWithErrorLogged(storageZk);

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

@@ -307,9 +307,6 @@
 import org.apache.helix.HelixManagerProperty;
 import org.apache.helix.InstanceType;
 import org.apache.helix.PropertyKey;
-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.manager.zk.ZKHelixAdmin;
 import org.apache.helix.manager.zk.ZKHelixManager;
 import org.apache.helix.manager.zk.ZNRecordSerializer;
@@ -362,7 +359,6 @@ public class VeniceHelixAdmin implements Admin, StoreCleaner {
   private final String kafkaSSLBootstrapServers;
   private final Map<String, AdminConsumerService> adminConsumerServices = new ConcurrentHashMap<>();
 
-  private static final int CONTROLLER_CLUSTER_NUMBER_OF_PARTITION = 1;
   private static final long CONTROLLER_CLUSTER_RESOURCE_EV_TIMEOUT_MS = TimeUnit.MINUTES.toMillis(5);
   private static final long CONTROLLER_CLUSTER_RESOURCE_EV_CHECK_DELAY_MS = 500;
   private static final long HELIX_RESOURCE_ASSIGNMENT_RETRY_INTERVAL_MS = 500;
@@ -7296,22 +7292,12 @@ private void createClusterIfRequired(String clusterName) {
         delayedTime);
     admin.addStateModelDef(clusterName, LeaderStandbySMD.name, LeaderStandbySMD.build());
 
-    admin.addResource(
-        controllerClusterName,
-        clusterName,
-        CONTROLLER_CLUSTER_NUMBER_OF_PARTITION,
-        LeaderStandbySMD.name,
-        IdealState.RebalanceMode.FULL_AUTO.toString(),
-        AutoRebalanceStrategy.class.getName());
-    IdealState idealState = admin.getResourceIdealState(controllerClusterName, clusterName);
-    // Use crush alg to allocate controller as well.
-    int controllerClusterReplica = config.getControllerClusterReplica();
-    idealState.setReplicas(String.valueOf(controllerClusterReplica));
-    idealState.setMinActiveReplicas(controllerClusterReplica);
-    idealState.setRebalancerClassName(DelayedAutoRebalancer.class.getName());
-    idealState.setRebalanceStrategy(CrushRebalanceStrategy.class.getName());
-    admin.setResourceIdealState(controllerClusterName, clusterName, idealState);
-    admin.rebalance(controllerClusterName, clusterName, controllerClusterReplica);
+    // 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);
+    }
   }
 
   public boolean updateIdealState(String clusterName, String resourceName, int minReplica) {