           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]
-  (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]))))
+(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)

                          (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)]

        last-active (atom false)        
        spouts (ArrayList. (map :object (vals task-datas)))
        rand (Random. (Utils/secureRandomLong))

        ;; 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.)

        pending (RotatingMap.
                 2 ;; microoptimize for performance of .size method
                 (reify RotatingMap$ExpiredCallback
