         event-handler (mk-task-receiver executor-data tuple-action-fn)
         has-ackers? (has-ackers? storm-conf)
         emitted-count (MutableLong. 0)
-        empty-emit-streak (MutableLong. 0)
-        
-        ;; the overflow buffer is used to ensure that spouts never block when emitting
-        ;; this ensures that the spout can always clear the incoming buffer (acks and fails), which
-        ;; prevents deadlock from occuring across the topology (e.g. Spout -> Bolt -> Acker -> Spout, and all
-        ;; buffers filled up)
-        ;; when the overflow buffer is full, spouts stop calling nextTuple until it's able to clear the overflow buffer
-        ;; this limits the size of the overflow buffer to however many tuples a spout emits in one call of nextTuple, 
-        ;; preventing memory issues
-        overflow-buffer (ConcurrentLinkedQueue.)]
+        empty-emit-streak (MutableLong. 0)]
    
     [(async-loop
       (fn []

                          (let [stream-id (.getSourceStreamId tuple)]
                            (condp = stream-id
                              Constants/SYSTEM_TICK_STREAM_ID (.rotate pending)
                              Constants/METRICS_TICK_STREAM_ID (metrics-tick executor-data (get task-datas task-id) tuple)
                              Constants/CREDENTIALS_CHANGED_STREAM_ID 
                                (let [task-data (get task-datas task-id)
                                      spout-obj (:object task-data)]

          receive-queue
          [[nil (TupleImpl. worker-context [interval] Constants/SYSTEM_TASK_ID Constants/METRICS_TICK_STREAM_ID)]]))))))

(defn metrics-tick
  ([executor-data task-data ^TupleImpl tuple overflow-buffer]
   (let [{:keys [interval->task->metric-registry ^WorkerTopologyContext worker-context]} executor-data
         interval (.getInteger tuple 0)
         task-id (:task-id task-data)
         name->imetric (-> interval->task->metric-registry (get interval) (get task-id))
         task-info (IMetricsConsumer$TaskInfo.
                     (hostname (:storm-conf executor-data))
                     (.getThisWorkerPort worker-context)
                     (:component-id executor-data)
                     task-id
                     (long (/ (System/currentTimeMillis) 1000))
                     interval)
         data-points (->> name->imetric
                          (map (fn [[name imetric]]
                                 (let [value (.getValueAndReset ^IMetric imetric)]
                                   (if value
                                     (IMetricsConsumer$DataPoint. name value)))))
                          (filter identity)
                          (into []))]
     (if (seq data-points)
       (task/send-unanchored task-data Constants/METRICS_STREAM_ID [task-info data-points] overflow-buffer))))
  ([executor-data task-data ^TupleImpl tuple]
    (metrics-tick executor-data task-data tuple nil)
    ))


(defn setup-ticks! [worker executor-data]
  (let [storm-conf (:storm-conf executor-data)
