ZOOKEEPER-3167: add an API and the corresponding CLI to get total count of recursive sub nodes under a specific path

- Thanks the original patch from [TyqITstudent ](https://github.com/TyqITstudent).
- the `getAllChildrenNumber` api supports `sync` and `async`, don't support `watch`.
  since` getChildren()` can get the number of the first level,to keep the api short and clean,don't use a boolean flag to let this api get the number of the first level child.
- the implements using the` parallelStream()` which have the almost `4x` speed up than `forEach`(I test),it will return at once even a millons of keys.
- `getAllChildrenNumber` doesn't have a `synchronized `lock, to avoid holding the `nodes` for a long time,so when too many concurrent writes, the number will be not very precise.
- about `getAllChildrenNumber` CLI:

    > [zk: localhost:2181(CONNECTED) 6] getAllChildrenNumber /zookeeper
    > 	2
    > 	[zk: localhost:2181(CONNECTED) 7] getAllChildrenNumber /zookeeper/quota
    > 	0

- there is no good place to document the new api, I create a seperate jira-[ZOOKEEPER-3259](https://issues.apache.org/jira/projects/ZOOKEEPER/issues/ZOOKEEPER-3259).

Author: maoling <maoling199210191@sina.com>

Reviewers: andor@apache.org

Closes #790 from maoling/ZOOKEEPER-3167

zookeeper-docs/src/main/resources/markdown/zookeeperStarted.md

@@ -153,6 +153,7 @@ From the shell, type `help` to get a listing of commands that can be executed fr
         set path data [version]
         delquota [-n|-b] path
         quit
+        getAllChildrenNumber path
         printwatches on|off
         create path data acl
         stat path [watch]

zookeeper-jute/src/main/resources/zookeeper.jute

@@ -142,6 +142,9 @@ module org.apache.zookeeper.proto {
         ustring path;
         boolean watch;
     }
+    class GetAllChildrenNumberRequest {
+        ustring path;
+    }
     class GetChildren2Request {
         ustring path;
         boolean watch;
@@ -206,6 +209,9 @@ module org.apache.zookeeper.proto {
     class GetChildrenResponse {
         vector<ustring> children;
     }
+    class GetAllChildrenNumberResponse {
+         int totalNumber;
+    }
     class GetChildren2Response {
         vector<ustring> children;
         org.apache.zookeeper.data.Stat stat;

zookeeper-server/src/main/java/org/apache/zookeeper/AsyncCallback.java

@@ -68,6 +68,20 @@ public interface AsyncCallback {
         public void processResult(int rc, String path, Object ctx, Stat stat);
     }
 
+    /*
+    *  This callback is used to get all children node number of the node.
+    * */
+    @InterfaceAudience.Public
+    interface AllChildrenNumberCallback extends AsyncCallback {
+        /**
+         *
+         * @param rc      The return code or the result of the call.
+         * @param ctx     Whatever context object that we passed to asynchronous calls.
+         * @param number  the number of children nodes under a specific path
+         */
+        public void processResult(int rc, String path, Object ctx, int number);
+    }
+
     /**
      * This callback is used to retrieve the data and stat of the node.
      */

zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java

@@ -51,6 +51,7 @@ import org.apache.jute.Record;
 import org.apache.zookeeper.AsyncCallback.ACLCallback;
 import org.apache.zookeeper.AsyncCallback.Children2Callback;
 import org.apache.zookeeper.AsyncCallback.ChildrenCallback;
+import org.apache.zookeeper.AsyncCallback.AllChildrenNumberCallback;
 import org.apache.zookeeper.AsyncCallback.Create2Callback;
 import org.apache.zookeeper.AsyncCallback.DataCallback;
 import org.apache.zookeeper.AsyncCallback.EphemeralsCallback;
@@ -77,6 +78,7 @@ import org.apache.zookeeper.proto.CreateResponse;
 import org.apache.zookeeper.proto.ExistsResponse;
 import org.apache.zookeeper.proto.GetACLResponse;
 import org.apache.zookeeper.proto.GetChildren2Response;
+import org.apache.zookeeper.proto.GetAllChildrenNumberResponse;
 import org.apache.zookeeper.proto.GetChildrenResponse;
 import org.apache.zookeeper.proto.GetDataResponse;
 import org.apache.zookeeper.proto.GetEphemeralsResponse;
@@ -562,6 +564,9 @@ public class ClientCnxn {
                     } else if (lcb.cb instanceof AsyncCallback.EphemeralsCallback) {
                         ((AsyncCallback.EphemeralsCallback) lcb.cb).processResult(lcb.rc,
                               lcb.ctx, null);
+                    } else if (lcb.cb instanceof AsyncCallback.AllChildrenNumberCallback) {
+                        ((AsyncCallback.AllChildrenNumberCallback) lcb.cb).processResult(lcb.rc,
+                                lcb.path, lcb.ctx, -1);
                     } else {
                         ((VoidCallback) lcb.cb).processResult(lcb.rc, lcb.path,
                                 lcb.ctx);
@@ -625,6 +630,14 @@ public class ClientCnxn {
                       } else {
                           cb.processResult(rc, clientPath, p.ctx, null);
                       }
+                  } else if (p.response instanceof GetAllChildrenNumberResponse) {
+                      AllChildrenNumberCallback cb = (AllChildrenNumberCallback) p.cb;
+                      GetAllChildrenNumberResponse rsp = (GetAllChildrenNumberResponse) p.response;
+                      if (rc == 0) {
+                          cb.processResult(rc, clientPath, p.ctx, rsp.getTotalNumber());
+                      } else {
+                          cb.processResult(rc, clientPath, p.ctx, -1);
+                      }
                   } else if (p.response instanceof GetChildren2Response) {
                       Children2Callback cb = (Children2Callback) p.cb;
                       GetChildren2Response rsp = (GetChildren2Response) p.response;

zookeeper-server/src/main/java/org/apache/zookeeper/ZooDefs.java

@@ -83,6 +83,8 @@ public class ZooDefs {
 
         public final int getEphemerals = 103;
 
+        public final int getAllChildrenNumber = 104;
+
         public final int createSession = -10;
 
         public final int closeSession = -11;

zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java

@@ -53,6 +53,8 @@ import org.apache.zookeeper.proto.GetACLRequest;
 import org.apache.zookeeper.proto.GetACLResponse;
 import org.apache.zookeeper.proto.GetChildren2Request;
 import org.apache.zookeeper.proto.GetChildren2Response;
+import org.apache.zookeeper.proto.GetAllChildrenNumberRequest;
+import org.apache.zookeeper.proto.GetAllChildrenNumberResponse;
 import org.apache.zookeeper.proto.GetChildrenRequest;
 import org.apache.zookeeper.proto.GetChildrenResponse;
 import org.apache.zookeeper.proto.GetDataRequest;
@@ -2668,6 +2670,62 @@ public class ZooKeeper implements AutoCloseable {
         getChildren(path, watch ? watchManager.defaultWatcher : null, cb, ctx);
     }
 
+    /**
+     * Synchronously gets all numbers of children nodes under a specific path
+     *
+     * @since 3.6.0
+     * @param path
+     * @return
+     * @throws KeeperException
+     * @throws InterruptedException
+     */
+    public int getAllChildrenNumber(final String path)
+            throws KeeperException, InterruptedException {
+
+        final String clientPath = path;
+        PathUtils.validatePath(clientPath);
+
+        final String serverPath = prependChroot(clientPath);
+
+        RequestHeader h = new RequestHeader();
+        h.setType(ZooDefs.OpCode.getAllChildrenNumber);
+        GetAllChildrenNumberRequest request = new GetAllChildrenNumberRequest(serverPath);
+        GetAllChildrenNumberResponse response = new GetAllChildrenNumberResponse();
+
+        ReplyHeader r = cnxn.submitRequest(h, request, response, null);
+        if (r.getErr() != 0) {
+            throw KeeperException.create(KeeperException.Code.get(r.getErr()),
+                    clientPath);
+        }
+        return response.getTotalNumber();
+    }
+
+    /**
+     * Asynchronously gets all numbers of children nodes under a specific path
+     *
+     * @since 3.6.0
+     * @param path
+     * @return
+     * @throws KeeperException
+     * @throws InterruptedException
+     */
+    public void getAllChildrenNumber(final String path, AsyncCallback.AllChildrenNumberCallback cb, Object ctx) {
+
+        final String clientPath = path;
+        PathUtils.validatePath(clientPath);
+
+        final String serverPath = prependChroot(clientPath);
+
+        RequestHeader h = new RequestHeader();
+        h.setType(ZooDefs.OpCode.getAllChildrenNumber);
+        GetAllChildrenNumberRequest request = new GetAllChildrenNumberRequest(serverPath);
+        GetAllChildrenNumberResponse response = new GetAllChildrenNumberResponse();
+
+        cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
+                clientPath, serverPath, ctx, null);
+    }
+
+
     /**
      * Synchronously gets all the ephemeral nodes  created by this session.
      *

zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeperMain.java

@@ -37,6 +37,7 @@ import java.util.NoSuchElementException;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.apache.zookeeper.cli.CliException;
 import org.apache.zookeeper.cli.CommandNotFoundException;
+import org.apache.zookeeper.cli.GetAllChildrenNumberCommand;
 import org.apache.zookeeper.cli.GetEphemeralsCommand;
 import org.apache.zookeeper.cli.MalformedCommandException;
 import org.slf4j.Logger;
@@ -124,7 +125,8 @@ public class ZooKeeperMain {
         new GetConfigCommand().addToMap(commandMapCli);
         new RemoveWatchesCommand().addToMap(commandMapCli);
         new GetEphemeralsCommand().addToMap(commandMapCli);
-        
+        new GetAllChildrenNumberCommand().addToMap(commandMapCli);
+
         // add all to commandMap
         for (Entry<String, CliCommand> entry : commandMapCli.entrySet()) {
             commandMap.put(entry.getKey(), entry.getValue().getOptionStr());

zookeeper-server/src/main/java/org/apache/zookeeper/cli/GetAllChildrenNumberCommand.java

@@ -0,0 +1,70 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.apache.zookeeper.cli;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.commons.cli.Parser;
+import org.apache.commons.cli.PosixParser;
+import org.apache.zookeeper.KeeperException;
+
+/**
+ * getAllChildrenNumber command for CLI
+ */
+public class GetAllChildrenNumberCommand extends CliCommand {
+    private static Options options = new Options();
+    private String[] args;
+
+    public GetAllChildrenNumberCommand() {
+        super("getAllChildrenNumber", "path");
+    }
+
+    @Override
+    public CliCommand parse(String[] cmdArgs) throws CliParseException {
+        Parser parser = new PosixParser();
+        CommandLine cl;
+        try {
+            cl = parser.parse(options, cmdArgs);
+        } catch (ParseException ex) {
+            throw new CliParseException(ex);
+        }
+        args = cl.getArgs();
+
+        return this;
+    }
+
+    @Override
+    public boolean exec() throws CliException {
+        if (args.length < 2) {
+            throw new MalformedCommandException(getUsageStr());
+        }
+
+        try {
+            String path = args[1];
+            int allChildrenNumber = zk.getAllChildrenNumber(path);
+
+            out.println(allChildrenNumber);
+        } catch (IllegalArgumentException ex) {
+            throw new MalformedPathException(ex.getMessage());
+        } catch (KeeperException | InterruptedException ex) {
+            throw new CliWrapperException(ex);
+        }
+
+        return false;
+    }
+}

zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java

@@ -703,6 +703,15 @@ public class DataTree {
         }
     }
 
+    public int getAllChildrenNumber(String path) {
+        //cull out these two keys:"", "/"
+        if ("/".equals(path)) {
+            return nodes.size() - 2;
+        }
+
+        return (int)nodes.keySet().parallelStream().filter(key -> key.startsWith(path + "/")).count();
+    }
+
     public Stat setACL(String path, List<ACL> acl, int version)
             throws KeeperException.NoNodeException {
         Stat stat = new Stat();

zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java

@@ -53,6 +53,8 @@ import org.apache.zookeeper.proto.GetACLRequest;
 import org.apache.zookeeper.proto.GetACLResponse;
 import org.apache.zookeeper.proto.GetChildren2Request;
 import org.apache.zookeeper.proto.GetChildren2Response;
+import org.apache.zookeeper.proto.GetAllChildrenNumberRequest;
+import org.apache.zookeeper.proto.GetAllChildrenNumberResponse;
 import org.apache.zookeeper.proto.GetChildrenRequest;
 import org.apache.zookeeper.proto.GetChildrenResponse;
 import org.apache.zookeeper.proto.GetDataRequest;
@@ -427,6 +429,24 @@ public class FinalRequestProcessor implements RequestProcessor {
                 rsp = new GetChildrenResponse(children);
                 break;
             }
+            case OpCode.getAllChildrenNumber: {
+                lastOp = "GETACN";
+                GetAllChildrenNumberRequest getAllChildrenNumberRequest = new
+                        GetAllChildrenNumberRequest();
+                ByteBufferInputStream.byteBuffer2Record(request.request,
+                        getAllChildrenNumberRequest);
+                path = getAllChildrenNumberRequest.getPath();
+                DataNode n = zks.getZKDatabase().getNode(path);
+                if (n == null) {
+                    throw new KeeperException.NoNodeException();
+                }
+                PrepRequestProcessor.checkACL(zks, request.cnxn, zks.getZKDatabase().aclForNode(n),
+                        ZooDefs.Perms.READ,
+                        request.authInfo, path, null);
+                int number = zks.getZKDatabase().getAllChildrenNumber(path);
+                rsp = new GetAllChildrenNumberResponse(number);
+                break;
+             }
             case OpCode.getChildren2: {
                 lastOp = "GETC";
                 GetChildren2Request getChildren2Request = new GetChildren2Request();

zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java

@@ -854,6 +854,7 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
             case OpCode.getData:
             case OpCode.getACL:
             case OpCode.getChildren:
+            case OpCode.getAllChildrenNumber:
             case OpCode.getChildren2:
             case OpCode.ping:
             case OpCode.setWatches:

zookeeper-server/src/main/java/org/apache/zookeeper/server/Request.java

@@ -145,6 +145,7 @@ public class Request {
         case OpCode.exists:
         case OpCode.getACL:
         case OpCode.getChildren:
+        case OpCode.getAllChildrenNumber:
         case OpCode.getChildren2:
         case OpCode.getData:
         case OpCode.getEphemerals:
@@ -168,6 +169,7 @@ public class Request {
         case OpCode.exists:
         case OpCode.getACL:
         case OpCode.getChildren:
+        case OpCode.getAllChildrenNumber:
         case OpCode.getChildren2:
         case OpCode.getData:
         case OpCode.getEphemerals:
@@ -229,6 +231,8 @@ public class Request {
             return "setACL";
         case OpCode.getChildren:
             return "getChildren";
+        case OpCode.getAllChildrenNumber:
+            return "getAllChildrenNumber";
         case OpCode.getChildren2:
             return "getChildren2";
         case OpCode.getEphemerals:

zookeeper-server/src/main/java/org/apache/zookeeper/server/TraceFormatter.java

@@ -59,6 +59,8 @@ public class TraceFormatter {
             return "setACL";
         case OpCode.getChildren:
             return "getChildren";
+        case OpCode.getAllChildrenNumber:
+            return "getAllChildrenNumber";
         case OpCode.getChildren2:
             return "getChildren2";
         case OpCode.getEphemerals:

zookeeper-server/src/main/java/org/apache/zookeeper/server/ZKDatabase.java

@@ -516,6 +516,14 @@ public class ZKDatabase {
         return dataTree.getChildren(path, stat, watcher);
     }
 
+    /*
+    * get all sub-children number of this node
+    * */
+    public int getAllChildrenNumber(String path)
+            throws KeeperException.NoNodeException {
+        return dataTree.getAllChildrenNumber(path);
+    }
+
     /**
      * check if the path is special or not
      * @param path the input path

zookeeper-server/src/test/java/org/apache/zookeeper/GetAllChildrenNumberTest.java

@@ -0,0 +1,112 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper;
+
+import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.test.ClientBase;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class GetAllChildrenNumberTest extends ClientBase {
+    private static final String BASE = "/getAllChildrenNumberTest";
+    private static final String BASE_EXT = BASE + "EXT";
+    private static final int PERSISTENT_CNT = 2;
+    private static final int EPHEMERAL_CNT = 3;
+
+    private ZooKeeper zk;
+
+    @Override
+    public void setUp() throws Exception {
+        super.setUp();
+
+        zk = createClient();
+        generatePaths(PERSISTENT_CNT, EPHEMERAL_CNT);
+    }
+
+    @Override
+    public void tearDown() throws Exception {
+        super.tearDown();
+
+        zk.close();
+    }
+
+    @Test
+    public void testGetAllChildrenNumberSync() throws KeeperException, InterruptedException {
+        //a bad case
+        try {
+            zk.getAllChildrenNumber(null);
+            Assert.fail("the path for getAllChildrenNumber must not be null.");
+        } catch (IllegalArgumentException e) {
+            //expected
+        }
+
+        Assert.assertEquals(EPHEMERAL_CNT, zk.getAllChildrenNumber(BASE + "/0"));
+        Assert.assertEquals(0, zk.getAllChildrenNumber(BASE + "/0/ephem0"));
+        Assert.assertEquals(0, zk.getAllChildrenNumber(BASE_EXT));
+        Assert.assertEquals(PERSISTENT_CNT + PERSISTENT_CNT * EPHEMERAL_CNT, zk.getAllChildrenNumber(BASE));
+        // 6(EPHEMERAL) + 2(PERSISTENT) + 3("/zookeeper,/zookeeper/quota,/zookeeper/config") + 1(BASE_EXT) + 1(BASE) = 13
+        Assert.assertEquals(13, zk.getAllChildrenNumber("/"));
+    }
+
+    @Test
+    public void testGetAllChildrenNumberAsync() throws IOException, KeeperException, InterruptedException {
+
+        final CountDownLatch doneProcessing = new CountDownLatch(1);
+
+        zk.getAllChildrenNumber("/", new AsyncCallback.AllChildrenNumberCallback() {
+            @Override
+            public void processResult(int rc, String path, Object ctx, int number) {
+                if (path == null) {
+                    Assert.fail((String.format("the path of getAllChildrenNumber was null.")));
+                }
+                Assert.assertEquals(13, number);
+                doneProcessing.countDown();
+            }
+        }, null);
+        long waitForCallbackSecs = 2L;
+        if (!doneProcessing.await(waitForCallbackSecs, TimeUnit.SECONDS)) {
+            Assert.fail(String.format("getAllChildrenNumber didn't callback within %d seconds",
+                    waitForCallbackSecs));
+        }
+    }
+
+    private void generatePaths(int persistantCnt, int ephemeralCnt)
+            throws KeeperException, InterruptedException {
+
+        zk.create(BASE, BASE.getBytes(), Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+        zk.create(BASE_EXT, BASE_EXT.getBytes(), Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+
+        for (int p = 0; p < persistantCnt; p++) {
+            String base = BASE + "/" + p;
+            zk.create(base, base.getBytes(), Ids.OPEN_ACL_UNSAFE,
+                    CreateMode.PERSISTENT);
+            for (int e = 0; e < ephemeralCnt; e++) {
+                String ephem = base + "/ephem" + e;
+                zk.create(ephem, ephem.getBytes(), Ids.OPEN_ACL_UNSAFE,
+                        CreateMode.EPHEMERAL);
+            }
+        }
+    }
+}

zookeeper-server/src/test/java/org/apache/zookeeper/server/DataTreeTest.java

@@ -342,4 +342,20 @@ public class DataTreeTest extends ZKTestCase {
         dt.deleteNode("/testApproximateDataSize", -1);
         Assert.assertEquals(dt.cachedApproximateDataSize(), dt.approximateDataSize());
     }
+
+    @Test
+    public void testGetAllChildrenNumber() throws Exception {
+        DataTree dt = new DataTree();
+        // create a node
+        dt.createNode("/all_children_test", new byte[20], null, -1, 1, 1, 1);
+        dt.createNode("/all_children_test/nodes", new byte[20], null, -1, 1, 1, 1);
+        dt.createNode("/all_children_test/nodes/node1", new byte[20], null, -1, 1, 1, 1);
+        dt.createNode("/all_children_test/nodes/node2", new byte[20], null, -1, 1, 1, 1);
+        dt.createNode("/all_children_test/nodes/node3", new byte[20], null, -1, 1, 1, 1);
+        Assert.assertEquals(4, dt.getAllChildrenNumber("/all_children_test"));
+        Assert.assertEquals(3, dt.getAllChildrenNumber("/all_children_test/nodes"));
+        Assert.assertEquals(0, dt.getAllChildrenNumber("/all_children_test/nodes/node1"));
+        //add these three init nodes:/zookeeper,/zookeeper/quota,/zookeeper/config,so the number is 8.
+        Assert.assertEquals( 8, dt.getAllChildrenNumber("/"));
+    }
 }