                       )
                     (reportError [this error]
                       (report-error error)
-                      ))
-                  )))
+                      )))))
+        (reset! open-or-prepare-was-called? true) 
         (log-message "Opened spout " component-id ":" (keys task-datas))
         
         (disruptor/consumer-started! (:receive-queue executor-data))

                       (reportError [this error]
                         (report-error error)
                         )))))
        (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)))
        
        (log-message "Opening spout " component-id ":" (keys task-datas))
        (doseq [[task-id task-data] task-datas
