                            (not= (global-amt track-id "transferred")                                 
                                  (global-amt track-id "processed"))
                            ))]
-        (while (waiting?)
+        (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"))

    
    
    (let [storm-id (common/get-storm-id state storm-name)]
      (while (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)
        (simulate-wait cluster-map))
      (when cleanup-state
        (doseq [spout (spout-objects spouts)]

                  supervisors
                  workers) ; because a worker may already be dead
        ]
    (while-timeout TEST-TIMEOUT-MS (not (every? (memfn waiting?) daemons))
      (Thread/sleep 10)
;;      (doseq [d daemons]
;;        (if-not ((memfn waiting?) d)
