                                       bolt-obj (:object task-data)]
                                   (when (instance? ICredentialsListener bolt-obj)
                                     (.setCredentials bolt-obj (.getValue tuple 0))))
-                              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)
                               (let [task-data (get task-datas task-id)
                                     ^IBolt bolt-obj (:object task-data)
                                     user-context (:user-context task-data)

                                    (stats/bolt-execute-tuple! executor-stats
                                                               (.getSourceComponent tuple)
                                                               (.getSourceStreamId tuple)
                                                               delta)))))))

        ;; the overflow buffer is used to ensure that bolts do not block when emitting
        ;; this ensures that the bolt can always clear the incoming messages, which
        ;; prevents deadlock from occurs across the topology
        ;; (e.g. Spout -> BoltA -> Splitter -> BoltB -> BoltA, and all
        ;; buffers filled up)
        ;; the overflow buffer is might gradually fill degrading the performance gradually
        ;; eventually running out of memory, but at least prevent live-locks/deadlocks.
        overflow-buffer (if (storm-conf TOPOLOGY-BOLTS-OUTGOING-OVERFLOW-BUFFER-ENABLE) (ConcurrentLinkedQueue.) nil)]
    
    ;; TODO: can get any SubscribedState objects out of the context now


        {:keys [storm-conf component-id worker-context transfer-fn report-error sampler
                open-or-prepare-was-called?]} executor-data
        rand (Random. (Utils/secureRandomLong))

        ;; the overflow buffer is used to ensure that bolts do not block when emitting
        ;; this ensures that the bolt can always clear the incoming messages, which
        ;; prevents deadlock from occurs across the topology
        ;; (e.g. Spout -> BoltA -> Splitter -> BoltB -> BoltA, and all
        ;; buffers filled up)
        ;; the overflow buffer is might gradually fill degrading the performance gradually
        ;; eventually running out of memory, but at least prevent live-locks/deadlocks.
        overflow-buffer (if (storm-conf TOPOLOGY-BOLTS-OUTGOING-OVERFLOW-BUFFER-ENABLE) (ConcurrentLinkedQueue.) nil)
        tuple-action-fn (fn [task-id ^TupleImpl tuple]
                          ;; synchronization needs to be done with a key provided by this bolt, otherwise:
                          ;; spout 1 sends synchronization (s1), dies, same spout restarts somewhere else, sends synchronization (s2) and incremental update. s2 and update finish before s1 -> lose the incremental update
