                  (int (storm-conf TOPOLOGY-MESSAGE-TIMEOUT-SECS))
                  (reify TimeCacheMap$ExpiredCallback
                    (expire [this msg-id [spout-id tuple-info start-time-ms]]
-                     (let [time-delta (if start-time-ms (time-delta-ms start-time-ms))]
-                       (.add event-queue #(fail-spout-msg spout storm-conf spout-id tuple-info time-delta task-stats)))
+                     (let [time-delta (time-delta-ms start-time-ms)]
+                       (.add event-queue #(fail-spout-msg spout storm-conf spout-id tuple-info time-delta task-stats sampler)))
                      )))
         send-spout-msg (fn [out-stream-id values message-id out-task-id]
                          (let [out-tasks (if out-task-id

           (let [tuple (.deserialize deserializer ser-msg)
                 id (.getValue tuple 0)
                 [spout-id tuple-finished-info start-time-ms] (.remove pending id)
                 time-delta (if start-time-ms (time-delta-ms start-time-ms))]
             (when spout-id
               (condp = (.getSourceStreamId tuple)
                 ACKER-ACK-STREAM-ID (.add event-queue #(ack-spout-msg spout storm-conf spout-id
                                                                       tuple-finished-info time-delta task-stats))
                 ACKER-FAIL-STREAM-ID (.add event-queue #(fail-spout-msg spout storm-conf spout-id
                                                                         tuple-finished-info time-delta task-stats))
                 )))
           ;; TODO: on failure, emit tuple to failure stream
           )))

                               (send-unanchored topology-context tasks-fn transfer-fn
                                                ACKER-FAIL-STREAM-ID [root]))
                             (let [delta (tuple-time-delta! tuple-start-times tuple)]
                               (when (sampler)
                                 (stats/bolt-failed-tuple! task-stats
                                                           (.getSourceComponent tuple)
                                                           (.getSourceStreamId tuple)
