     
     
     (let [storm-id (common/get-storm-id state storm-name)]
-      (while (not (every? exhausted? (spout-objects spouts)))
+      (while-timeout TEST-TIMEOUT-MS (not (every? exhausted? (spout-objects spouts)))
         (simulate-wait cluster-map))
 
       (.killTopologyWithOpts (:nimbus cluster-map) storm-name (doto (KillOptions.) (.set_wait_secs 0)))
-      (while (.assignment-info state storm-id nil)
+      (while-timeout TEST-TIMEOUT-MS (.assignment-info state storm-id nil)
         (simulate-wait cluster-map))
       (when cleanup-state
         (doseq [spout (spout-objects spouts)]

    (rmr t)
    ))


(defn wait-until-cluster-waiting
  "Wait until the cluster is idle. Should be used with time simulation."

                           (not= (global-amt track-id "transferred")                                 
                                 (global-amt track-id "processed"))
                           ))]
        (while-timeout TEST-TIMEOUT-MS (waiting?)
          ;; (println "Spout emitted: " (global-amt track-id "spout-emitted"))
          ;; (println "Processed: " (global-amt track-id "processed"))
          ;; (println "Transferred: " (global-amt track-id "transferred"))
