diff --git a/zookeeper-common/src/main/java/org/apache/zookeeper/cli/SetAclCommand.java b/zookeeper-common/src/main/java/org/apache/zookeeper/cli/SetAclCommand.java
index d2cfc0d3526..9d1b4606e1e 100644
--- a/zookeeper-common/src/main/java/org/apache/zookeeper/cli/SetAclCommand.java
+++ b/zookeeper-common/src/main/java/org/apache/zookeeper/cli/SetAclCommand.java
@@ -19,12 +19,17 @@
 
 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 boolean exec() throws CliException {
             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());
diff --git a/zookeeper-common/src/test/java/org/apache/zookeeper/ZooKeeperTest.java b/zookeeper-common/src/test/java/org/apache/zookeeper/ZooKeeperTest.java
index b0ac07fc2d5..f8be441dafc 100644
--- a/zookeeper-common/src/test/java/org/apache/zookeeper/ZooKeeperTest.java
+++ b/zookeeper-common/src/test/java/org/apache/zookeeper/ZooKeeperTest.java
@@ -566,4 +566,29 @@ public void testLsrNonexistantZnodeCommand() throws Exception {
             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()));
+    }
 }