make zookeeper cluster used for transactional state customizable

Nathan Marz · Nathan Marz · commit 9c3a4b761410 · 2012-01-27T15:58:44.000-08:00
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
@@ -32,6 +32,8 @@ drpc.port: 3772
 drpc.invocations.port: 3773
 
 transactional.zookeeper.root: "/transactional"
+transactional.zookeeper.servers: null
+transactional.zookeeper.port: null
 
 ### supervisor.* configs are for node supervisors
 # Define the amount of workers that can be run on this machine. Each worker is assigned a port to use for communication
diff --git a/src/jvm/backtype/storm/Config.java b/src/jvm/backtype/storm/Config.java
@@ -371,6 +371,18 @@ public class Config extends HashMap<String, Object> {
      */
     public static String TRANSACTIONAL_ZOOKEEPER_ROOT="transactional.zookeeper.root";
     
+    /**
+     * The list of zookeeper servers in which to keep the transactional state. If null (which is default),
+     * will use storm.zookeeper.servers
+     */
+    public static String TRANSACTIONAL_ZOOKEEPER_SERVERS="transactional.zookeeper.servers";
+
+    /**
+     * The port to use to connect to the transactional zookeeper servers. If null (which is default),
+     * will use storm.zookeeper.port
+     */
+    public static String TRANSACTIONAL_ZOOKEEPER_PORT="transactional.zookeeper.port";
+    
     /**
      * The number of threads that should be used by the zeromq context in each worker process.
      */
diff --git a/src/jvm/backtype/storm/transactional/state/TransactionalState.java b/src/jvm/backtype/storm/transactional/state/TransactionalState.java
@@ -35,7 +35,10 @@ protected TransactionalState(Map conf, String id, Map componentConf, String subr
                               .get(Config.TOPOLOGY_KRYO_REGISTER));
             }
             String rootDir = conf.get(Config.TRANSACTIONAL_ZOOKEEPER_ROOT) + "/" + id + "/" + subroot;
-            CuratorFramework initter = Utils.newCurator(conf);
+            List<String> servers = (List<String>) getWithBackup(conf, Config.TRANSACTIONAL_ZOOKEEPER_SERVERS, Config.STORM_ZOOKEEPER_SERVERS);
+            Object port = getWithBackup(conf, Config.TRANSACTIONAL_ZOOKEEPER_PORT, Config.STORM_ZOOKEEPER_PORT);
+            Object sessionTimeout = conf.get(Config.STORM_ZOOKEEPER_SESSION_TIMEOUT);
+            CuratorFramework initter = Utils.newCurator(servers, port, sessionTimeout);
             try {
                 initter.create().creatingParentsIfNeeded().forPath(rootDir);
             } catch(KeeperException.NodeExistsException e)  {
@@ -44,7 +47,7 @@ protected TransactionalState(Map conf, String id, Map componentConf, String subr
             
             initter.close();
                                     
-            _curator = Utils.newCurator(conf, rootDir);
+            _curator = Utils.newCurator(servers, port, sessionTimeout, rootDir);
             _ser = new KryoValuesSerializer(conf);
             _des = new KryoValuesDeserializer(conf);
         } catch (Exception e) {
@@ -111,4 +114,10 @@ public Object getData(String path) {
     public void close() {
         _curator.close();
     }
+    
+    private Object getWithBackup(Map amap, Object primary, Object backup) {
+        Object ret = amap.get(primary);
+        if(ret==null) return amap.get(backup);
+        return ret;
+    }
 }
diff --git a/src/jvm/backtype/storm/utils/Utils.java b/src/jvm/backtype/storm/utils/Utils.java
@@ -1,6 +1,5 @@
 package backtype.storm.utils;
 
-import backtype.storm.Config;
 import backtype.storm.generated.ComponentCommon;
 import backtype.storm.generated.ComponentObject;
 import backtype.storm.generated.StormTopology;
@@ -206,15 +205,15 @@ public static long randomLong() {
         return UUID.randomUUID().getLeastSignificantBits();
     }
     
-    public static CuratorFramework newCurator(Map conf, String root) {
+    public static CuratorFramework newCurator(List<String> servers, Object port, Object sessionTimeout, String root) {
         List<String> serverPorts = new ArrayList<String>();
-        for(String zkServer: (List<String>) conf.get(Config.STORM_ZOOKEEPER_SERVERS)) {
-            serverPorts.add(zkServer + ":" + Utils.getInt(conf.get(Config.STORM_ZOOKEEPER_PORT)));
+        for(String zkServer: (List<String>) servers) {
+            serverPorts.add(zkServer + ":" + Utils.getInt(port));
         }
         String zkStr = StringUtils.join(serverPorts, ",") + root; 
         try {
             CuratorFramework ret =  CuratorFrameworkFactory.newClient(zkStr,
-                                        Utils.getInt(conf.get(Config.STORM_ZOOKEEPER_SESSION_TIMEOUT)),
+                                        Utils.getInt(sessionTimeout),
                                         15000, new RetryNTimes(5, 1000));
             ret.start();
             return ret;
@@ -223,8 +222,8 @@ public static CuratorFramework newCurator(Map conf, String root) {
         }
     }
 
-    public static CuratorFramework newCurator(Map conf) {
-        return newCurator(conf, "");
+    public static CuratorFramework newCurator(List<String> servers, Object port, Object sessionTimeout) {
+        return newCurator(servers, port, sessionTimeout, "");
     }
     
     /**
