       (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.")
-    ledger))
+    (->BookKeeperLog bk-client ledger-handle)))
 
 
 (defn default-state 

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

  (let [bk-client (bookkeeper peer-opts)
        ensemble-size (arg-or-default :onyx.bookkeeper/ledger-ensemble-size peer-opts)
        quorum-size (arg-or-default :onyx.bookkeeper/ledger-quorum-size peer-opts)
        ledger-handle (create-ledger bk-client ensemble-size quorum-size digest-type (password peer-opts))
        slot-id (state/peer-slot-id event)
        new-ledger-id (.getId ledger-handle)] 
    (>!! outbox-ch
         {:fn :assign-bookkeeper-log-id
          :args {:job-id job-id
