           sync-callback (fn [& ignored] (.add event-manager this))
           assignment-versions @(:assignment-versions supervisor)
           {assignments-snapshot :assignments versions :versions}  (assignments-snapshot 
-                                                                   storm-cluster-state sync-callback 
+                                                                   storm-cluster-state sync-callback
                                                                    assignment-versions)
           storm-code-map (read-storm-code-locations assignments-snapshot)
           downloaded-storm-ids (set (read-downloaded-storm-ids conf))

      (log-debug "Downloaded storm ids: " downloaded-storm-ids)
      (log-debug "All assignment: " all-assignment)
      (log-debug "New assignment: " new-assignment)
      
      ;; download code first
      ;; This might take awhile
      ;;   - should this be done separately from usual monitoring?

                        (keys keepers))
           (zipmap (vals new-worker-ids) (keys new-worker-ids))
           ))

    ;; check storm topology code dir exists before launching workers
    (doseq [[port assignment] reassign-executors]
      (let [storm-cluster-state (:storm-cluster-state supervisor)
            downloaded-storm-ids (set (read-downloaded-storm-ids conf))
            storm-id (:storm-id assignment)
            assignment-info (.assignment-info-with-version storm-cluster-state storm-id nil)
            storm-code-map (read-storm-code-locations assignment-info)
            master-code-dir (if (contains? storm-code-map :data) (storm-code-map :data))
            stormroot (supervisor-stormdist-root conf storm-id)]
        (if-not (or (downloaded-storm-ids storm-id) (.exists (File. stormroot)))
          (download-storm-code conf storm-id master-code-dir))
        ))

    (wait-for-workers-launch
     conf
     (dofor [[port assignment] reassign-executors]
