                        (:failed curr)
                        (do
                          (.remove pending id)
-                         (acker-emit-direct @output-collector
+                         (acker-emit-direct output-collector
                                             (:spout-task curr)
                                             ACKER-FAIL-STREAM-ID
                                             [id]
                                             ))
                        ))
-               (.ack ^OutputCollector @output-collector tuple)
+               (.ack output-collector tuple)
                ))
       (^void cleanup [this]
-        (.cleanup @pending))
+        (.cleanup (.getObject pending)))
       )))
 
 (defn -init []

(ns backtype.storm.daemon.acker
  (:import [backtype.storm.task OutputCollector TopologyContext IBolt])
  (:import [backtype.storm.tuple Tuple Fields])
  (:import [backtype.storm.utils TimeCacheMap])
  (:import [java.util List Map])
  (:use [backtype.storm config util log])
  (:gen-class

                 (cond (= 0 (:val curr))
                       (do
                         (.remove pending id)
                         (acker-emit-direct output-collector
                                            (:spout-task curr)
                                            ACKER-ACK-STREAM-ID
                                            [id]
