    (FileUtils/writeByteArrayToFile (File. (master-stormconf-path stormroot)) (Utils/serialize storm-conf))
    ))
 
-(defn- read-storm-topology [conf storm-id]
-  (let [stormroot (master-stormdist-root conf storm-id)]
+(defn- read-storm-topology [storage storm-id]
+  (let [stormroot (nimbus-storage-stormdist-root storm-id)]
     (Utils/deserialize
-      (FileUtils/readFileToByteArray
-        (File. (master-stormcode-path stormroot))
-        ))))
+      (IOUtils/toByteArray
+        (.open storage (master-stormcode-path stormroot))))))
 
 (declare compute-executor->component)
 

        storm-base (.storm-base (:storm-cluster-state nimbus) storm-id nil)
        component->executors (:component->executors storm-base)
        storm-conf (read-storm-conf storage conf storm-id)
        topology (read-storm-topology conf storm-id)
        task->component (storm-task-info topology storm-conf)]
    (->> (storm-task-info topology storm-conf)
         reverse-map

  (let [conf (:conf nimbus)
        storage (:storage nimbus)
        executors (compute-executors nimbus storm-id)
        topology (read-storm-topology storage storm-id)
        storm-conf (read-storm-conf storage conf storm-id)
        task->component (storm-task-info topology storm-conf)
        executor->component (into {} (for [executor executors
