           ^ISupervisor isupervisor (:isupervisor supervisor)
           ^LocalState local-state (:local-state supervisor)
           sync-callback (fn [& ignored] (.add event-manager this))
-          storm-code-map (read-storm-code-locations storm-cluster-state sync-callback)
+          assignments-snapshot (assignments-snapshot storm-cluster-state sync-callback)
+          storm-code-map (read-storm-code-locations assignments-snapshot)
           downloaded-storm-ids (set (read-downloaded-storm-ids conf))
           all-assignment (read-assignments
-                           storm-cluster-state
-                           (:assignment-id supervisor)
-                           sync-callback)
+                           assignments-snapshot
+                           (:assignment-id supervisor))
           new-assignment (->> all-assignment
                               (filter-key #(.confirmAssigned isupervisor %)))
           assigned-storm-ids (assigned-storm-ids-from-port-assignments new-assignment)

  (shutdown-all-workers [this])
  )


(defn- read-my-executors [storm-cluster-state storm-id assignment-id callback]
  (let [assignment (.assignment-info storm-cluster-state storm-id callback)
        my-executors (filter (fn [[_ [node _]]] (= node assignment-id))
                               (:executor->node+port assignment))
        port-executors (apply merge-with
                          concat
                          (for [[executor [_ port]] my-executors]

               ;; need to cast to int b/c it might be a long (due to how yaml parses things)
               ;; doall is to avoid serialization/deserialization problems with lazy seqs
               [(Integer. port) (LocalAssignment. storm-id (doall executors))]
               ))))


(defn- read-assignments
  "Returns map from port to struct containing :storm-id and :executors"
  [assignments-snapshot assignment-id]
  (->> (dofor [sid (keys assignments-snapshot)] (read-my-executors assignments-snapshot sid assignment-id))
       (apply merge-with (fn [& ignored] (throw-runtime "Should not have multiple topologies assigned to one port")))))

(defn- read-storm-code-locations
  [assignments-snapshot]
  (map-val :master-code-dir assignments-snapshot))

(defn- read-downloaded-storm-ids [conf]
  (map #(java.net.URLDecoder/decode %) (read-dir-contents (supervisor-stormdist-root conf)))
