   (:require [onyx.log.curator :as curator]
             [taoensso.timbre :refer [info error warn trace fatal] :as timbre]
             [com.stuartsierra.component :as component]
-            [clojure.core.async :refer  [chan timeout thread go >! <! <!! >!! alts!! close!]]
+            [clojure.core.async :refer [chan timeout thread go >! <! <!! >!! alts!! close! poll!]]
+            [clojure.core.async.impl.protocols :refer [closed?]]
             [onyx.compression.nippy :as nippy]
             [onyx.extensions :as extensions]
             [onyx.monitoring.measurements :refer [emit-latency-value emit-latency]]


(def HandleWriteCallback
  (reify AsyncCallback$AddCallback
    (addComplete [this rc lh entry-id ack-fn]
      (if (= rc (BKException$Code/OK))
        (ack-fn)
        ;; TODO: should restart peer, see https://github.com/onyx-platform/onyx/issues/390
        (warn "Unable to complete async write to bookkeeper. BookKeeper exception code:" rc)))))

(defn compaction-transition 
  "Transitions to a new compacted ledger, plus a newly created ledger created

                 :task-id task-id
                 :slot-id slot-id
                 :ledger-id new-ledger-id}})
    (loop []
      (let [exit? (nil? (first (alts!! [kill-ch task-kill-ch] :default true)))
            not-added? (not= new-ledger-id (last (event->ledger-ids event)))]
        (cond exit? 
              (info "Exiting assign-bookkeeper-log-id-spin early as peer has been reassigned.")
              not-added? 
              (do
                (info "New ledger id has not been published yet. Backing off.")
                (Thread/sleep (arg-or-default :onyx.bookkeeper/ledger-id-written-back-off peer-opts))
                (recur))))))) 

(defmethod state-extensions/initialize-log :bookkeeper [log-type {:keys [onyx.core/peer-opts] :as event}] 
  (let [bk-client (bookkeeper peer-opts)
