                         (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]

          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 
                                                                   assignment-versions)
          storm-code-map (read-storm-code-locations assignments-snapshot)
          downloaded-storm-ids (set (read-downloaded-storm-ids conf))

    ;; Downloading to permanent location is atomic
    (let [tmproot (str (supervisor-tmp-dir conf) file-path-separator (uuid))
          stormroot (supervisor-stormdist-root conf storm-id)]
      (log-message "Downloading code for storm id "
                   storm-id
                   " from "
                   master-code-dir)
      (FileUtils/forceMkdir (File. tmproot))

      (Utils/downloadFromMaster conf (master-stormjar-path master-code-dir) (supervisor-stormjar-path tmproot))
      (Utils/downloadFromMaster conf (master-stormcode-path master-code-dir) (supervisor-stormcode-path tmproot))
      (Utils/downloadFromMaster conf (master-stormconf-path master-code-dir) (supervisor-stormconf-path tmproot))
      (extract-dir-from-jar (supervisor-stormjar-path tmproot) RESOURCES-SUBDIR tmproot)
      (if-not (.exists (File. stormroot))
        (FileUtils/moveDirectory (File. tmproot) (File. stormroot))
        (FileUtils/deleteDirectory (File. tmproot)))
      (setup-storm-code-dir conf (read-supervisor-storm-conf conf storm-id) stormroot)
      (log-message "Finished downloading code for storm id "
                   storm-id
                   " from "
                   master-code-dir)
    ))

(defn write-log-metadata-to-yaml-file! [storm-id port data conf]
