ZOOKEEPER-1908: setAcl should be have a recursive function

Let setAcl support recursive option, a UT is included.

Author: Reid Chan <reidchan@apache.org>

Reviewers: fangmin@apache.org, andor@apache.org

Closes #650 from Reidddddd/ZOOKEEPER-1908 and squashes the following commits:

2d0b33fd [Reid Chan] ZOOKEEPER-1908: Address review comments
9c9c1bb9 [Reid Chan] ZOOKEEPER-1908: Add unit test for setAcl recursive
9ec3d837 [Reid Chan] ZOOKEEPER-1908: setAcl should be have a recursive function

src/java/main/org/apache/zookeeper/cli/SetAclCommand.java

@@ -19,12 +19,17 @@ package org.apache.zookeeper.cli;
 
 import java.util.List;
 import org.apache.commons.cli.*;
+import org.apache.zookeeper.AsyncCallback.StringCallback;
 import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZKUtil;
 import org.apache.zookeeper.data.ACL;
 import org.apache.zookeeper.data.Stat;
 
 /**
- * setAcl command for cli
+ * setAcl command for cli.
+ * Available options are s for printing znode's stats, v for set version of znode(s), R for
+ * recursive setting. User can combine v and R options together, but not s and R considering the
+ * number of znodes could be large.
  */
 public class SetAclCommand extends CliCommand {
 
@@ -35,10 +40,11 @@ public class SetAclCommand extends CliCommand {
     {
         options.addOption("s", false, "stats");
         options.addOption("v", true, "version");
+        options.addOption("R", false, "recursive");
     }
 
     public SetAclCommand() {
-        super("setAcl", "[-s] [-v version] path acl");
+        super("setAcl", "[-s] [-v version] [-R] path acl");
     }
 
     @Override
@@ -69,9 +75,22 @@ public class SetAclCommand extends CliCommand {
             version = -1;
         }
         try {
-            Stat stat = zk.setACL(path, acl, version);
-            if (cl.hasOption("s")) {
-                new StatPrinter(out).print(stat);
+            if (cl.hasOption("R")) {
+                ZKUtil.visitSubTreeDFS(zk, path, false, new StringCallback() {
+                    @Override
+                    public void processResult(int rc, String p, Object ctx, String name) {
+                        try {
+                            zk.setACL(p, acl, version);
+                        } catch (KeeperException | InterruptedException e) {
+                            out.print(e.getMessage());
+                        }
+                    }
+                });
+            } else {
+                Stat stat = zk.setACL(path, acl, version);
+                if (cl.hasOption("s")) {
+                    new StatPrinter(out).print(stat);
+                }
             }
         } catch (IllegalArgumentException ex) {
             throw new MalformedPathException(ex.getMessage());

src/java/test/org/apache/zookeeper/ZooKeeperTest.java

@@ -566,4 +566,29 @@ public class ZooKeeperTest extends ClientBase {
             Assert.assertEquals(KeeperException.Code.NONODE, ((KeeperException)e.getCause()).code());
         }
     }
+
+    @Test
+    public void testSetAclRecursive() throws Exception {
+        final ZooKeeper zk = createClient();
+        final byte[] EMPTY = new byte[0];
+
+        zk.setData("/", EMPTY, -1);
+        zk.create("/a", EMPTY, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+        zk.create("/a/b", EMPTY, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+        zk.create("/a/b/c", EMPTY, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+        zk.create("/a/d", EMPTY, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+        zk.create("/e", EMPTY, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+
+        ZooKeeperMain zkMain = new ZooKeeperMain(zk);
+        String setAclCommand = "setAcl -R /a world:anyone:r";
+        zkMain.cl.parseCommand(setAclCommand);
+        Assert.assertFalse(zkMain.processZKCmd(zkMain.cl));
+
+        Assert.assertEquals(Ids.READ_ACL_UNSAFE, zk.getACL("/a", new Stat()));
+        Assert.assertEquals(Ids.READ_ACL_UNSAFE, zk.getACL("/a/b", new Stat()));
+        Assert.assertEquals(Ids.READ_ACL_UNSAFE, zk.getACL("/a/b/c", new Stat()));
+        Assert.assertEquals(Ids.READ_ACL_UNSAFE, zk.getACL("/a/d", new Stat()));
+        // /e is unset, its acl should remain the same.
+        Assert.assertEquals(Ids.OPEN_ACL_UNSAFE, zk.getACL("/e", new Stat()));
+    }
 }