         (every? (memfn sleeping?) system-threads)
         ))))
 
-(defn- fail-spout-msg [^ISpout spout storm-conf msg-id tuple-info time-delta task-stats]
+(defn- fail-spout-msg [^ISpout spout storm-conf msg-id tuple-info time-delta task-stats sampler]
   (log-message "Failing message " msg-id ": " tuple-info)
   (.fail spout msg-id)
-  (when time-delta
+  (when (sampler)
     (stats/spout-failed-tuple! task-stats (:stream tuple-info) time-delta)
     ))
 
-(defn- ack-spout-msg [^ISpout spout storm-conf msg-id tuple-info time-delta task-stats]
+(defn- ack-spout-msg [^ISpout spout storm-conf msg-id tuple-info time-delta task-stats sampler]
   (when (= true (storm-conf TOPOLOGY-DEBUG))
     (log-message "Acking message " msg-id))
   (.ack spout msg-id)
-  (when time-delta
+  (when (sampler)
     (stats/spout-acked-tuple! task-stats (:stream tuple-info) time-delta)
     ))
 

                               (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)
