 (defn password [peer-opts]
   (.getBytes (arg-or-default :onyx.bookkeeper/ledger-password peer-opts)))
 
+(defrecord BookKeeperLog [client ledger-handle])
+
 ;; TODO: add zk-timeout for bookkeeper
 (defmethod state-extensions/initialise-log :bookkeeper [log-type {:keys [onyx.core/replica onyx.core/peer-opts
                                                                          onyx.core/job-id onyx.core/task-id

            state
            ledger-ids))) 

(defmethod state-extensions/playback-log-entries org.apache.bookkeeper.client.LedgerHandle 
  [log {:keys [onyx.core/windows onyx.core/replica onyx.core/peer-opts onyx.core/job-id onyx.core/task-id] :as event} state]
  (let [slot-id (state/peer-slot-id event)
        ;; Don't play back the final ledger id because we just created it
        previous-ledger-ids (butlast (get-in @replica [:state-logs job-id task-id slot-id]))
        bk-client (bookkeeper peer-opts)
        _ (info "Playing back ledgers for" job-id task-id slot-id "ledger-ids" previous-ledger-ids)]
    (try
      (playback-ledgers bk-client peer-opts state previous-ledger-ids windows)
      (finally
        (.close bk-client)))))

(defmethod state-extensions/close-log org.apache.bookkeeper.client.LedgerHandle
  [log event] 
  (.close ^LedgerHandle log))

(defmethod state-extensions/store-log-entry org.apache.bookkeeper.client.LedgerHandle 
  [log event entry]
  (let [start-time (System/currentTimeMillis)]
    (info "Writing entry " entry)
    ;; TODO: make add entry async and use acking
    (.addEntry ^LedgerHandle log (nippy/freeze entry {}))
    (info "TOOK: " (- (System/currentTimeMillis) start-time)))
  ;; TODO: write out batch end (don't apply until batch end found in others)
  )

      (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))) 
    (info "Ledger id published.")
    (->BookKeeperLog bk-client ledger-handle)))


(defn default-state 
