                        (reportError [this error]
                          (report-error error)
                          )))))
+        (reset! open-or-prepare-was-called? true)        
         (log-message "Prepared bolt " component-id ":" (keys task-datas))          
 
         (let [receive-queue (:receive-queue executor-data)

      (fn []
        ;; If topology was started in inactive state, don't call (.open spout) until it's activated first.
        (sleep-until-true (fn [] @(:storm-active-atom executor-data)))
        (reset! open-or-prepare-was-called? true)

        
        (log-message "Opening spout " component-id ":" (keys task-datas))
        (doseq [[task-id task-data] task-datas

    [(async-loop
      (fn []
        (sleep-until-true (fn [] @(:storm-active-atom executor-data)))
        
        (log-message "Preparing bolt " component-id ":" (keys task-datas))
        (doseq [[task-id task-data] task-datas
                :let [^IBolt bolt-obj (:object task-data)
