[zookeeper] remove hardcoded zookeeper session timeout (apache#2889)

*Motivation* If zookeeper hostname resolution takes longer time (e.g. more than 10s). both initing cluster and starting broker will fail due to session timeout. *Changes* remove hardcoded zookeeper session timeout and make them take settings from configuration or arguments.
amaliujia · Oct 30, 2018 · 0a7c64b · 0a7c64b
1 parent 95f4385
commit 0a7c64b
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 3 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java
@@ -80,6 +80,11 @@ private static class Arguments {
                 "--zookeeper" }, description = "Local ZooKeeper quorum connection string", required = true)
         private String zookeeper;
 
+        @Parameter(names = {
+            "--zookeeper-session-timeout-ms"
+        }, description = "Local zookeeper session timeout ms")
+        private int zkSessionTimeoutMillis = 30000;
+
         @Parameter(names = { "-gzk",
                 "--global-zookeeper" }, description = "Global ZooKeeper quorum connection string", required = false, hidden = true)
         private String globalZookeeper;
@@ -126,13 +131,16 @@ public static void main(String[] args) throws Exception {
         log.info("Setting up cluster {} with zk={} configuration-store ={}", arguments.cluster, arguments.zookeeper,
                 arguments.configurationStore);
         ZooKeeperClientFactory zkfactory = new ZookeeperClientFactoryImpl();
-        ZooKeeper localZk = zkfactory.create(arguments.zookeeper, SessionType.ReadWrite, 30000).get();
-        ZooKeeper configStoreZk = zkfactory.create(arguments.configurationStore, SessionType.ReadWrite, 30000).get();
+        ZooKeeper localZk = zkfactory.create(
+            arguments.zookeeper, SessionType.ReadWrite, arguments.zkSessionTimeoutMillis).get();
+        ZooKeeper configStoreZk = zkfactory.create(
+            arguments.configurationStore, SessionType.ReadWrite, arguments.zkSessionTimeoutMillis).get();
 
         // Format BookKeeper metadata
         ServerConfiguration bkConf = new ServerConfiguration();
         bkConf.setLedgerManagerFactoryClass(HierarchicalLedgerManagerFactory.class);
         bkConf.setZkServers(arguments.zookeeper);
+        bkConf.setZkTimeout(arguments.zkSessionTimeoutMillis);
         if (localZk.exists("/ledgers", false) == null // only format if /ledgers doesn't exist
                 && !BookKeeperAdmin.format(bkConf, false /* interactive */, false /* force */)) {
             throw new IOException("Failed to initialize BookKeeper metadata");

diff --git a/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/GlobalZooKeeperCache.java b/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/GlobalZooKeeperCache.java
@@ -65,7 +65,7 @@ public void start() throws IOException {
 
         // Initial session creation with global ZK must work
         try {
-            ZooKeeper newSession = zkFuture.get(10, TimeUnit.SECONDS);
+            ZooKeeper newSession = zkFuture.get(zkSessionTimeoutMillis, TimeUnit.MILLISECONDS);
             // Register self as a watcher to receive notification when session expires and trigger a new session to be
             // created
             newSession.register(this);