 (defn- bucket-consumer-shutdown-handler!
   "Cleans up resources on consumer shutdown based on observing changes to
    the consumer state."
-  [old-state {:keys [instance-id status active empty-signal ch consumer-tag]}]
+  [old-state {:keys [instance-id status active ^BlockingQueue empty-signal ch consumer-tag]}]
 
   (cond
     ;; Let the shutdown thread know (via blockingqueue) when all the active

          broadcast! #(send-broadcast! conn broadcast-exchange peer-id %)

          ; core stuff that gets passed around
          distributor {:options options
                       :default-buckets default-buckets
                       :peer-id peer-id 
                       :state-atom state-atom
                       :broadcast! broadcast!}

          broadcast-consumer (start-broadcast-consumer! conn broadcast-exchange
                                                        #(handle-broadcast! distributor %1 %2))

  [state-atom & {:keys [force-stop]}]

    (loop []
      (let [{:keys [status ^BlockingQueue empty-signal]} (swap! state-atom stop-swap force-stop)]
        (when (not= status :stopped)
          (log/debugf "[%s] waiting for empty signal" (:instance-id @state-atom))
          (.take empty-signal) ;; wait for a message indicating we should try again
