zk client config update and bugfix for ZKMetadataClientDriver

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

@@ -75,6 +75,7 @@ public abstract class AbstractConfiguration<T extends AbstractConfiguration>
     // Zookeeper Parameters
     protected static final String ZK_TIMEOUT = "zkTimeout";
     protected static final String ZK_SERVERS = "zkServers";
+    protected static final String ZK_RETRY_BACKOFF_MAX_RETRIES = "zkRetryBackoffMaxRetries";
 
     // Ledger Manager
     protected static final String LEDGER_MANAGER_TYPE = "ledgerManagerType";
@@ -345,6 +346,27 @@ public T setZkTimeout(int zkTimeout) {
         return getThis();
     }
 
+    /**
+     * Get zookeeper client backoff max retry times.
+     *
+     * @return zk backoff max retry times.
+     */
+    public int getZkRetryBackoffMaxRetries() {
+        return getInt(ZK_RETRY_BACKOFF_MAX_RETRIES, Integer.MAX_VALUE);
+    }
+
+    /**
+     * Set zookeeper client backoff max retry times.
+     *
+     * @param maxRetries
+     *          backoff max retry times
+     * @return server configuration.
+     */
+    public T setZkRetryBackoffMaxRetries(int maxRetries) {
+        setProperty(ZK_RETRY_BACKOFF_MAX_RETRIES, Integer.toString(maxRetries));
+        return getThis();
+    }
+
     /**
      * Set Ledger Manager Type.
      *

bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataBookieDriver.java

@@ -56,7 +56,7 @@ public synchronized MetadataBookieDriver initialize(ServerConfiguration conf,
             conf,
             statsLogger.scope(BOOKIE_SCOPE),
             new BoundExponentialBackoffRetryPolicy(conf.getZkRetryBackoffStartMs(),
-                        conf.getZkRetryBackoffMaxMs(), Integer.MAX_VALUE),
+                        conf.getZkRetryBackoffMaxMs(), conf.getZkRetryBackoffMaxRetries()),
             Optional.empty());
         this.serverConf = conf;
         this.statsLogger = statsLogger;

bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataClientDriver.java

@@ -66,7 +66,7 @@ public synchronized MetadataClientDriver initialize(ClientConfiguration conf,
             new BoundExponentialBackoffRetryPolicy(
                 conf.getZkTimeout(),
                 conf.getZkTimeout(),
-                0),
+                conf.getZkRetryBackoffMaxRetries()),
             optionalCtx);
         this.statsLogger = statsLogger;
         this.clientConf = conf;

bookkeeper-server/src/test/java/org/apache/bookkeeper/conf/TestBKConfiguration.java

@@ -60,6 +60,10 @@ public static ServerConfiguration newServerConfiguration() {
         confReturn.setAllocatorPoolingPolicy(PoolingPolicy.UnpooledHeap);
         confReturn.setProperty(DbLedgerStorage.WRITE_CACHE_MAX_SIZE_MB, 4);
         confReturn.setProperty(DbLedgerStorage.READ_AHEAD_CACHE_MAX_SIZE_MB, 4);
+        /**
+         * if testcase has zk error,just try 0 time for fast running
+         */
+        confReturn.setZkRetryBackoffMaxRetries(0);
         setLoopbackInterfaceAndAllowLoopback(confReturn);
         return confReturn;
     }
@@ -89,6 +93,10 @@ public static ServerConfiguration setLoopbackInterfaceAndAllowLoopback(ServerCon
     public static ClientConfiguration newClientConfiguration() {
         ClientConfiguration clientConfiguration = new ClientConfiguration();
         clientConfiguration.setTLSEnabledProtocols("TLSv1.2,TLSv1.1");
+        /**
+         * if testcase has zk error,just try 0 time for fast running
+         */
+        clientConfiguration.setZkRetryBackoffMaxRetries(0);
         return clientConfiguration;
     }
 }

stream/server/src/main/java/org/apache/bookkeeper/stream/server/service/RegistrationServiceProvider.java

@@ -66,7 +66,7 @@ public RegistrationServiceProvider(ServerConfiguration bkServerConf,
         this.bkZkRetryPolicy = new BoundExponentialBackoffRetryPolicy(
             bkServerConf.getZkRetryBackoffStartMs(),
             bkServerConf.getZkRetryBackoffMaxMs(),
-            Integer.MAX_VALUE);
+            bkServerConf.getZkRetryBackoffMaxRetries());
         this.regExecutor = Executors.newSingleThreadScheduledExecutor(
             new ThreadFactoryBuilder().setNameFormat("registration-service-provider-scheduler").build());
         ClientConfiguration clientConfiguration = new ClientConfiguration(bkServerConf);