             [baldr.core :refer (baldr-writer)]
             [uswitch.bifrost.util :refer (close-channels)]
             [uswitch.bifrost.async :refer (observable-chan)]
-            [metrics.meters :refer (meter mark! defmeter)])
+            [metrics.meters :refer (meter mark! defmeter)]
+            [metrics.counters :refer (defcounter inc!)])
   (:import [java.util.zip GZIPOutputStream]))
 
 (defn- get-available-topics

                        (timeout rotation-interval)))))
    message-ch))

(defn safe-zookeeper-consumer
  "Repeatedly tries to construct a ZooKeeper consumer. Will retry every
  15s and log errors. Returns a channel that contains the obtained

    (go-loop
     []
     (if-let [c (get-consumer)]
       (do
         (inc! zookeeper-consumers)
         c)
       (do (<! (timeout (* 15 1000)))
           (recur))))))

