ISSUE #263: Enable request rate limiter in zookeeper client

Descriptions of the changes in this PR:

Eanble request rate limiter in zookeeper client. The rate limiter logic is in place, but is never enabled. This change is to expose the settings in configurations and enable them in zookeeper client at both client and server.

Author: Sijie Guo <sijie@apache.org>

Reviewers: Enrico Olivelli <None>, Jia Zhai <None>

This closes #264 from sijie/interface_syncs, closes #263

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java

@@ -27,6 +27,7 @@
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_READ_ENTRY;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_READ_ENTRY_BYTES;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_RECOVERY_ADD_ENTRY;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_SCOPE;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.JOURNAL_SCOPE;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.LD_INDEX_SCOPE;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.LD_LEDGER_SCOPE;
@@ -165,6 +166,7 @@ public class Bookie extends BookieCriticalThread {
             new ThreadFactoryBuilder().setNameFormat("BookieStateService-%d").build());
 
     // Expose Stats
+    private final StatsLogger statsLogger;
     private final Counter writeBytes;
     private final Counter readBytes;
     // Bookie Operation Latency Stats
@@ -649,6 +651,7 @@ public Bookie(ServerConfiguration conf)
     public Bookie(ServerConfiguration conf, StatsLogger statsLogger)
             throws IOException, KeeperException, InterruptedException, BookieException {
         super("Bookie-" + conf.getBookiePort());
+        this.statsLogger = statsLogger;
         this.zkAcls = ZkUtils.getACLs(conf);
         this.bookieRegistrationPath = conf.getZkAvailableBookiesPath() + "/";
         this.bookieReadonlyRegistrationPath =
@@ -1205,6 +1208,8 @@ public void process(WatchedEvent event) {
                 .watchers(watchers)
                 .operationRetryPolicy(new BoundExponentialBackoffRetryPolicy(conf.getZkRetryBackoffStartMs(),
                         conf.getZkRetryBackoffMaxMs(), Integer.MAX_VALUE))
+                .requestRateLimit(conf.getZkRequestRateLimit())
+                .statsLogger(this.statsLogger.scope(BOOKIE_SCOPE))
                 .build();
     }
 

bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/AbstractConfiguration.java

@@ -59,6 +59,7 @@ public abstract class AbstractConfiguration extends CompositeConfiguration {
     protected final static String LEDGER_MANAGER_TYPE = "ledgerManagerType";
     protected final static String LEDGER_MANAGER_FACTORY_CLASS = "ledgerManagerFactoryClass";
     protected final static String ZK_LEDGERS_ROOT_PATH = "zkLedgersRootPath";
+    protected final static String ZK_REQUEST_RATE_LIMIT = "zkRequestRateLimit";
     protected final static String AVAILABLE_NODE = "available";
     protected final static String REREPLICATION_ENTRY_BATCH_SIZE = "rereplicationEntryBatchSize";
 
@@ -193,6 +194,25 @@ public String getZkLedgersRootPath() {
         return getString(ZK_LEDGERS_ROOT_PATH, "/ledgers");
     }
 
+    /**
+     * Get zookeeper access request rate limit.
+     *
+     * @return zookeeper access request rate limit.
+     */
+    public double getZkRequestRateLimit() {
+        return getDouble(ZK_REQUEST_RATE_LIMIT, 0);
+    }
+
+    /**
+     * Set zookeeper access request rate limit.
+     *
+     * @param rateLimit
+     *          zookeeper access request rate limit.
+     */
+    public void setZkRequestRateLimit(double rateLimit) {
+        setProperty(ZK_REQUEST_RATE_LIMIT, rateLimit);
+    }
+
     /**
      * Are z-node created with strict ACLs
      *