refactor reading of topology and topology config from supervisor cache

Nathan Marz · Nathan Marz · commit f7fa84ab3226 · 2012-01-04T23:11:42.000-08:00
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
@@ -68,3 +68,4 @@ topology.max.spout.pending: null
 topology.state.synchronization.timeout.secs: 60
 topology.stats.sample.rate: 0.05
 topology.fall.back.on.java.serialization: true
+topology.worker.childopts: nil
diff --git a/src/clj/backtype/storm/config.clj b/src/clj/backtype/storm/config.clj
@@ -128,6 +128,18 @@
 (defn ^LocalState supervisor-state [conf]
   (LocalState. (str (supervisor-local-dir conf) "/localstate")))
 
+(defn read-supervisor-storm-conf [conf storm-id]
+  (let [stormroot (supervisor-stormdist-root conf storm-id)
+        conf-path (supervisor-stormconf-path stormroot)
+        topology-path (supervisor-stormcode-path stormroot)]
+    (merge conf (Utils/deserialize (FileUtils/readFileToByteArray (File. conf-path))))
+    ))
+
+(defn read-supervisor-topology [conf storm-id]
+  (let [stormroot (supervisor-stormdist-root conf storm-id)
+        topology-path (supervisor-stormcode-path stormroot)]
+    (Utils/deserialize (FileUtils/readFileToByteArray (File. topology-path)))
+    ))
 
 (defn worker-root
   ([conf]
diff --git a/src/clj/backtype/storm/daemon/worker.clj b/src/clj/backtype/storm/daemon/worker.clj
@@ -24,14 +24,6 @@
             assignment))
     ))
 
-(defn- read-storm-cache [conf storm-id]
-  (let [stormroot (supervisor-stormdist-root conf storm-id)
-        conf-path (supervisor-stormconf-path stormroot)
-        topology-path (supervisor-stormcode-path stormroot)]
-    [(merge conf (Utils/deserialize (FileUtils/readFileToByteArray (File. conf-path))))
-     (Utils/deserialize (FileUtils/readFileToByteArray (File. topology-path)))]
-    ))
-
 (defn do-heartbeat [conf worker-id port storm-id task-ids]
   (.put (worker-state conf worker-id)
         LS-WORKER-HEARTBEAT
@@ -97,7 +89,8 @@
         ;; do this here so that the worker process dies if this fails
         ;; it's important that worker heartbeat to supervisor ASAP when launching so that the supervisor knows it's running (and can move on)
         _ (heartbeat-fn)
-        [storm-conf topology] (read-storm-cache conf storm-id)        
+        storm-conf (read-supervisor-storm-conf conf storm-id)
+        topology (read-supervisor-topology conf storm-id)
         event-manager (event/event-manager true)
         
         task->component (storm-task-info storm-cluster-state storm-id)
