Support custom ZooKeeper ACLs when creating a cluster

helix-core/src/main/java/org/apache/helix/HelixAdmin.java

@@ -45,6 +45,7 @@
 import org.apache.helix.model.ResourceConfig;
 import org.apache.helix.model.StateModelDefinition;
 import org.apache.helix.model.OperationCheckResult;
+import org.apache.zookeeper.data.ACL;
 
 /*
  * Helix cluster management
@@ -110,6 +111,34 @@ public interface HelixAdmin {
    */
   boolean addCluster(String clusterName, boolean recreateIfExists);
 
+  /**
+   * Add a cluster whose metadata store nodes are created with the given ACLs
+   * @param clusterName
+   * @param recreateIfExists If the cluster already exists, it will delete it and recreate
+   * @param acl ACLs applied to the cluster root node ("/{clusterName}") and to every cluster
+   *            metadata node created underneath it by this call. If null or empty, the default ACL
+   *            of the underlying metadata store client is used, making this equivalent to
+   *            {@link #addCluster(String, boolean)}. ZooKeeper does not propagate ACLs to children,
+   *            so nodes created after this call (resources, instances, live instances, ...) are
+   *            NOT covered and keep the client default ACL. The ACL is only applied when the nodes
+   *            are created by this call; the ACL of a pre-existing cluster is left untouched unless
+   *            recreateIfExists is true.
+   *            <p>
+   *            The supplied ACL must grant the calling client CREATE on the root, otherwise cluster
+   *            creation fails part way through and leaves an incomplete cluster behind. Deployments
+   *            running a server-side ACL provider that assigns ACLs on create may ignore this
+   *            argument entirely.
+   * @return true if successfully created, or if cluster already exists
+   * @throws UnsupportedOperationException if a non-empty ACL is supplied and the implementation
+   *         does not support custom ACLs
+   */
+  default boolean addCluster(String clusterName, boolean recreateIfExists, List<ACL> acl) {
+    if (acl == null || acl.isEmpty()) {
+      return addCluster(clusterName, recreateIfExists);
+    }
+    throw new UnsupportedOperationException("addCluster with ACL is not implemented.");
+  }
+
   /**
    * Add a cluster and also add this cluster as a resource group in the super cluster
    * @param clusterName

helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java

@@ -118,6 +118,7 @@
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.Op;
 import org.apache.zookeeper.OpResult;
+import org.apache.zookeeper.data.ACL;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -1644,6 +1645,11 @@ public boolean addCluster(String clusterName) {
 
   @Override
   public boolean addCluster(String clusterName, boolean recreateIfExists) {
+    return addCluster(clusterName, recreateIfExists, null);
+  }
+
+  @Override
+  public boolean addCluster(String clusterName, boolean recreateIfExists, List<ACL> acl) {
     logger.info("Add cluster {}.", clusterName);
     String root = "/" + clusterName;
 
@@ -1657,7 +1663,11 @@ public boolean addCluster(String clusterName, boolean recreateIfExists) {
       }
     }
     try {
-      _zkClient.createPersistent(root, true);
+      if (acl == null || acl.isEmpty()) {
+        _zkClient.createPersistent(root, true);
+      } else {
+        _zkClient.createPersistent(root, true, acl);
+      }
     } catch (Exception e) {
       // some other process might have created the cluster
       if (_zkClient.exists(root)) {
@@ -1667,7 +1677,7 @@ public boolean addCluster(String clusterName, boolean recreateIfExists) {
       return false;
     }
     try {
-      createZKPaths(clusterName);
+      createZKPaths(clusterName, acl);
     } catch (Exception e) {
       logger.error("Error creating cluster:" + clusterName, e);
       return false;
@@ -1676,49 +1686,69 @@ public boolean addCluster(String clusterName, boolean recreateIfExists) {
     return true;
   }
 
-  private void createZKPaths(String clusterName) {
+  private void createZKPaths(String clusterName, List<ACL> acl) {
     String path;
 
     // IDEAL STATE
-    _zkClient.createPersistent(PropertyPathBuilder.idealState(clusterName));
+    createPersistent(PropertyPathBuilder.idealState(clusterName), false, acl);
     // CONFIGURATIONS
     path = PropertyPathBuilder.clusterConfig(clusterName);
-    _zkClient.createPersistent(path, true);
+    createPersistent(path, true, acl);
     _zkClient.writeData(path, new ZNRecord(clusterName));
     path = PropertyPathBuilder.instanceConfig(clusterName);
-    _zkClient.createPersistent(path);
+    createPersistent(path, false, acl);
     path = PropertyPathBuilder.resourceConfig(clusterName);
-    _zkClient.createPersistent(path);
+    createPersistent(path, false, acl);
     path = PropertyPathBuilder.customizedStateConfig(clusterName);
-    _zkClient.createPersistent(path);
+    createPersistent(path, false, acl);
     // PROPERTY STORE
     path = PropertyPathBuilder.propertyStore(clusterName);
-    _zkClient.createPersistent(path);
+    createPersistent(path, false, acl);
     // LIVE INSTANCES
-    _zkClient.createPersistent(PropertyPathBuilder.liveInstance(clusterName));
+    createPersistent(PropertyPathBuilder.liveInstance(clusterName), false, acl);
     // MEMBER INSTANCES
-    _zkClient.createPersistent(PropertyPathBuilder.instance(clusterName));
+    createPersistent(PropertyPathBuilder.instance(clusterName), false, acl);
     // External view
-    _zkClient.createPersistent(PropertyPathBuilder.externalView(clusterName));
+    createPersistent(PropertyPathBuilder.externalView(clusterName), false, acl);
     // State model definition
-    _zkClient.createPersistent(PropertyPathBuilder.stateModelDef(clusterName));
+    createPersistent(PropertyPathBuilder.stateModelDef(clusterName), false, acl);
 
     // controller
-    _zkClient.createPersistent(PropertyPathBuilder.controller(clusterName));
+    createPersistent(PropertyPathBuilder.controller(clusterName), false, acl);
     path = PropertyPathBuilder.controllerHistory(clusterName);
     final ZNRecord emptyHistory = new ZNRecord(PropertyType.HISTORY.toString());
     final List<String> emptyList = new ArrayList<String>();
     emptyHistory.setListField(clusterName, emptyList);
-    _zkClient.createPersistent(path, emptyHistory);
+    createPersistent(path, emptyHistory, acl);
 
     path = PropertyPathBuilder.controllerMessage(clusterName);
-    _zkClient.createPersistent(path);
+    createPersistent(path, false, acl);
 
     path = PropertyPathBuilder.controllerStatusUpdate(clusterName);
-    _zkClient.createPersistent(path);
+    createPersistent(path, false, acl);
 
     path = PropertyPathBuilder.controllerError(clusterName);
-    _zkClient.createPersistent(path);
+    createPersistent(path, false, acl);
+  }
+
+  /**
+   * Creates a persistent node, applying the given ACL when one is supplied. A null or empty ACL
+   * falls back to the ZkClient default, preserving the behavior of clusters created without ACLs.
+   */
+  private void createPersistent(String path, boolean createParents, List<ACL> acl) {
+    if (acl == null || acl.isEmpty()) {
+      _zkClient.createPersistent(path, createParents);
+    } else {
+      _zkClient.createPersistent(path, createParents, acl);
+    }
+  }
+
+  private void createPersistent(String path, Object data, List<ACL> acl) {
+    if (acl == null || acl.isEmpty()) {
+      _zkClient.createPersistent(path, data);
+    } else {
+      _zkClient.createPersistent(path, data, acl);
+    }
   }
 
   @Override