            (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))]
+                 time-delta (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))
+                                                                       tuple-finished-info time-delta task-stats sampler))
                  ACKER-FAIL-STREAM-ID (.add event-queue #(fail-spout-msg spout storm-conf spout-id
-                                                                         tuple-finished-info time-delta task-stats))
+                                                                         tuple-finished-info time-delta task-stats sampler))
                  )))
            ;; 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 delta
                                 (stats/bolt-failed-tuple! task-stats
                                                           (.getSourceComponent tuple)
                                                           (.getSourceStreamId tuple)

                                                  ACKER-ACK-STREAM-ID [root (bit-xor id ack-val)])
                                 ))
                             (let [delta (tuple-time-delta! tuple-start-times tuple)]
                               (when (sampler)
                                 (stats/bolt-acked-tuple! task-stats
                                                          (.getSourceComponent tuple)
                                                          (.getSourceStreamId tuple)
