                           (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/METRICS_TICK_STREAM_ID (metrics-tick executor-data (get task-datas task-id) tuple overflow-buffer)
                               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]
  (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 setup-ticks! [worker executor-data]
  (let [storm-conf (:storm-conf executor-data)

        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)]
   
    [(async-loop
      (fn []
