ZOOKEEPER-3301:Enforce the quota limit

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

@@ -975,6 +975,10 @@ property, when available, is noted below.
     commit log is triggered.
     Does not affect the limit defined by *flushDelay*.
     Default is 1000.
+* *enforceQuota* :
+    (Java system property: **zookeeper.enforceQuota**)
+    enforce the quota check. Enable this option to use the [quota feature](http://zookeeper.apache.org/doc/current/zookeeperQuotas.html).
+    the default value:true.
 
 * *requestThrottleLimit* :
     (Java system property: **zookeeper.request_throttle_max_requests**)

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

@@ -42,7 +42,7 @@ ZooKeeper -server host:port cmd args
 	create [-s] [-e] [-c] [-t ttl] path [data] [acl]
 	delete [-v version] path
 	deleteall path
-	delquota [-n|-b] path
+	delquota [-n|-b|-N|-B] path
 	get [-s] [-w] path
 	getAcl [-s] path
 	getAllChildrenNumber path
@@ -57,7 +57,7 @@ ZooKeeper -server host:port cmd args
 	removewatches path [-c|-d|-a] [-l]
 	set [-s] [-v version] path data
 	setAcl [-s] [-v version] [-R] path acl
-	setquota -n|-b val path
+	setquota -n|-b|-N|-B val path
 	stat [-w] path
 	sync path
 	version
@@ -187,6 +187,11 @@ Delete the quota under a path
 [zkshell: 2] listquota /quota_test
 	absolute path is /zookeeper/quota/quota_test/zookeeper_limits
 	quota for /quota_test does not exist.
+[zkshell: 3] delquota -n /c1
+[zkshell: 4] delquota -N /c2
+[zkshell: 5] delquota -b /c3
+[zkshell: 6] delquota -B /c4
+
 ```
 ## get
 Get the data of the specific path
@@ -281,10 +286,10 @@ Showing the history about the recent 11 commands that you have executed
 Listing the quota of one path
 
 ```bash
-[zkshell: 1] listquota /quota_test
-	absolute path is /zookeeper/quota/quota_test/zookeeper_limits
-	Output quota for /quota_test count=2,bytes=-1
-	Output stat for /quota_test count=4,bytes=0
+[zkshell: 1] listquota /c1
+             absolute path is /zookeeper/quota/c1/zookeeper_limits
+             Output quota for /c1 count=-1,bytes=-1=;byteHardLimit=-1;countHardLimit=2
+             Output stat for /c1 count=4,bytes=0
 ```
 
 ## ls
@@ -497,6 +502,26 @@ Set the quota in one path.
 [zkshell: 23] set /brokers "I_love_zookeeper"
 # Notice:don't have a hard constraint,just log the warning info
 	WARN  [CommitProcWorkThread-7:DataTree@379] - Quota exceeded: /brokers bytes=4206 limit=5
+
+# -N count Hard quota
+[zkshell: 3] create /c1
+Created /c1
+[zkshell: 4] setquota -N 2 /c1
+[zkshell: 5] listquota /c1
+absolute path is /zookeeper/quota/c1/zookeeper_limits
+Output quota for /c1 count=-1,bytes=-1=;byteHardLimit=-1;countHardLimit=2
+Output stat for /c1 count=2,bytes=0
+[zkshell: 6] create /c1/ch-3
+Count Quota has exceeded : /c1/ch-3
+
+# -B byte Hard quota
+[zkshell: 3] create /c2
+[zkshell: 4] setquota -B 4 /c2
+[zkshell: 5] set /c2 "foo"
+[zkshell: 6] set /c2 "foo-bar"
+Bytes Quota has exceeded : /c2
+[zkshell: 7] get /c2
+foo
 ```
 
 ## stat

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

@@ -67,6 +67,9 @@ according to specific circumstances.
 
 - Users cannot set the quota on the path under **/zookeeper/quota**
 
+- The quota supports the soft and hard quota. The soft quota just logs the warning info when exceeding the quota, but the hard quota
+also throws a QuotaExceededException. When setting soft and hard quota on the same path, the hard quota has the priority.
+
 <a name="Listing+Quotas"></a>
 
 ### Listing Quotas

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

@@ -148,6 +148,10 @@ public static KeeperException create(Code code) {
             return new SessionClosedRequireAuthException();
         case REQUESTTIMEOUT:
             return new RequestTimeoutException();
+        case COUNTQUOTAEXCEEDED:
+            return new CountQuotaExceededException();
+        case BYTESQUOTAEXCEEDED:
+            return new ByteQuotaExceededException();
         case THROTTLEDOP:
             return new ThrottledOpException();
         case OK:
@@ -407,6 +411,10 @@ public enum Code implements CodeDeprecated {
          *  but client is not configured with SASL authentication or configuted with SASL but failed
          *  (i.e. wrong credential used.). */
         SESSIONCLOSEDREQUIRESASLAUTH(-124),
+        /** Exceeded the count quota that was set on the path.*/
+        COUNTQUOTAEXCEEDED(-125),
+        /** Exceeded the bytes quota that was set on the path.*/
+        BYTESQUOTAEXCEEDED(-126),
         /** Operation was throttled and not executed at all. This error code indicates that zookeeper server
          *  is under heavy load and can't process incoming requests at full speed; please retry with back off.
          */
@@ -501,6 +509,10 @@ static String getCodeMessage(Code code) {
             return "Reconfig is disabled";
         case SESSIONCLOSEDREQUIRESASLAUTH:
             return "Session closed because client failed to authenticate";
+        case COUNTQUOTAEXCEEDED:
+            return "Count Quota has exceeded";
+        case BYTESQUOTAEXCEEDED:
+            return "Bytes Quota has exceeded";
         case THROTTLEDOP:
             return "Op throttled due to high load";
         default:
@@ -948,6 +960,43 @@ public RequestTimeoutException() {
 
     }
 
+
+    @InterfaceAudience.Public
+    public static class QuotaExceededException extends KeeperException {
+        public QuotaExceededException(Code code) {
+            super(code);
+        }
+
+        public QuotaExceededException(Code code, String path) {
+            super(code, path);
+        }
+    }
+
+    /**
+     * @see Code#COUNTQUOTAEXCEEDED
+     */
+    @InterfaceAudience.Public
+    public static class CountQuotaExceededException extends QuotaExceededException {
+        public CountQuotaExceededException() {
+            super(Code.COUNTQUOTAEXCEEDED);
+        }
+        public CountQuotaExceededException(String path) {
+            super(Code.COUNTQUOTAEXCEEDED, path);
+        }
+    }
+
+    /**
+     * @see Code#BYTESQUOTAEXCEEDED
+     */
+    @InterfaceAudience.Public
+    public static class ByteQuotaExceededException extends QuotaExceededException {
+        public ByteQuotaExceededException() {
+            super(Code.BYTESQUOTAEXCEEDED);
+        }
+        public ByteQuotaExceededException(String path) {
+            super(Code.BYTESQUOTAEXCEEDED, path);
+        }
+    }
     /**
      * @see Code#THROTTLEDOP
      */

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

@@ -48,9 +48,19 @@ public class Quotas {
      * return the quota path associated with this
      * prefix
      * @param path the actual path in zookeeper.
-     * @return the limit quota path
+     * @return the quota path
      */
     public static String quotaPath(String path) {
+        return quotaZookeeper + path;
+    }
+
+    /**
+     * return the limit quota path associated with this
+     * prefix
+     * @param path the actual path in zookeeper.
+     * @return the limit quota path
+     */
+    public static String limitPath(String path) {
         return quotaZookeeper + path + "/" + limitNode;
     }
 
@@ -64,4 +74,13 @@ public static String statPath(String path) {
         return quotaZookeeper + path + "/" + statNode;
     }
 
+    /**
+     * return the real path associated with this
+     * quotaPath.
+     * @param quotaPath the quotaPath which's started with /zookeeper/quota
+     * @return the real path associated with this quotaPath.
+     */
+    public static String trimQuotaPath(String quotaPath) {
+        return quotaPath.substring(quotaZookeeper.length());
+    }
 }