     (doseq [[task-id task-data] task-datas
             :let [^IBolt bolt-obj (:object task-data)
                   tasks-fn (:tasks-fn task-data)
+                  user-context (:user-context task-data)
                   bolt-emit (fn [stream anchors values task]
                               (let [out-tasks (if task
                                                 (tasks-fn task stream values)

                                (or out-tasks [])))]]
      (.prepare bolt-obj
                storm-conf
                (:user-context task-data)
                (OutputCollector.
                  (reify IOutputCollector
                     (emit [this stream anchors values]

                                               ACKER-FAIL-STREAM-ID
                                               [root]))
                       (let [delta (tuple-time-delta! tuple-start-times tuple)]
                         (task/apply-hooks user-context .boltFail (BoltFailInfo. tuple delta))
                         (when (sampler)
                           (stats/bolt-failed-tuple! executor-stats
                                                     (.getSourceComponent tuple)
