         backoff-period 10
         commit-loop-ch (start-commit-loop! commit-ch log checkpoint-key)
         ledgers-root-path (or (:bookkeeper/zookeeper-ledgers-root-path task-map)
-                               (log-zk/ledgers-path (:onyx/id peer-opts)))
+                              (log-zk/ledgers-path (:onyx/id peer-opts)))
         zookeeper-addr (:bookkeeper/zookeeper-addr task-map)
         zookeeper-timeout 60000
         bookkeeper-throttle 30000
         client (obk/bookkeeper zookeeper-addr ledgers-root-path zookeeper-timeout bookkeeper-throttle)
         ledger-id (:bookkeeper/ledger-id task-map)
-        password (or (:bookkeeper/password-bytes task-map) (throw (Exception. ":bookkeeper/password-bytes must be supplied")))
-        no-recovery? (:bookkeeper/no-recovery? task-map)
+        password (or (:bookkeeper/password-bytes task-map) 
+                     (throw (Exception. ":bookkeeper/password-bytes must be supplied")))
         deserializer-fn (kw->fn (:bookkeeper/deserializer-fn task-map))
         digest (digest-type (:bookkeeper/digest-type task-map))
-        ledger-handle (if no-recovery? 
-                        (obk/open-ledger-no-recovery client ledger-id digest password)
-                        (obk/open-ledger client ledger-id digest password) )
+        read-fn (if (:bookkeeper/no-recovery? task-map)
+                  no-recovery-read-ledger-entries!
+                  read-ledger-entries!)
         producer-ch (thread
                       (try
-                        (let [exit (loop [last-acked (:largest checkpointed)]
-                                     (read-ledger-entries! client ledger-handle read-ch deserializer-fn last-acked max-id no-recovery? backoff-period)
+                        (let [exit (loop [last-acked (inc (:largest checkpointed))]
+                                     (read-fn client ledger-id digest password read-ch deserializer-fn backoff-period last-acked max-id)
                                      :finished)]
                           (if-not (= exit :shutdown)
                             (>!! read-ch (t/input (random-uuid) :done))))

    (throw (Exception. "Restarted task, however it was already completed for this job.
                       This is currently unhandled."))))

(defn read-ledger-chunk! [ledger-handle deserializer-fn ledger-id read-ch start end]
  (loop [entries (.readEntries ledger-handle start end)
         ledger-entry (.nextElement entries)] 
    (let [segment {:entry-id (.getEntryId ^LedgerEntry ledger-entry) 
                   :ledger-id ledger-id
                   :value (deserializer-fn (.getEntry ^LedgerEntry ledger-entry))}] 
      (>!! read-ch (t/input (random-uuid) segment))
      (if (.hasMoreElements entries)


(def read-chunk-size 100)

(defn read-ledger-entries! 
  [client ledger-id digest password read-ch deserializer-fn backoff-period start end]
  (let [ledger-handle (obk/open-ledger client ledger-id digest password)
        last-confirmed (.getLastAddConfirmed ledger-handle)
        bounded-end (min end last-confirmed)
        chunks (partition-all 2 1 (range (dec start) bounded-end read-chunk-size))
        _ (info "Starting final read: " start end last-confirmed bounded-end (vec chunks))]
    (run! (fn [[s e]]
              (read-ledger-chunk! ledger-handle deserializer-fn read-ch (inc s) (or e bounded-end)))
            chunks)))

(defn no-recovery-read-ledger-entries! 
  [client ledger-id digest password read-ch deserializer-fn backoff-period start end]
  (let [ledger-handle (obk/open-ledger-no-recovery client ledger-id digest password)
        last-confirmed (.getLastAddConfirmed ledger-handle)
        bounded-end (min end last-confirmed)
        chunks (partition-all 2 1 (range (dec start) bounded-end read-chunk-size))]
    (run! (fn [[s e]]
              (read-ledger-chunk! ledger-handle deserializer-fn read-ch (inc s) (or e bounded-end)))
            chunks)
    (if (.isClosed client ledger-id)
      (read-ledger-entries! client ledger-id digest password read-ch deserializer-fn backoff-period (inc bounded-end) end)
      (do
        (Thread/sleep 1000)
        (recur client ledger-id digest password read-ch deserializer-fn backoff-period (inc bounded-end) end)))))

(defn inject-read-ledgers-resources
  [{:keys [onyx.core/task-map onyx.core/log onyx.core/task-id onyx.core/pipeline onyx.core/peer-opts] :as event} lifecycle]
  (when-not (= 1 (:onyx/max-peers task-map))
    (throw (ex-info "Read log tasks must set :onyx/max-peers 1" task-map)))
  (let [; decrement because we are going to store this as a checkpoint and then recover
        start-id (dec (or (:bookkeeper/ledger-start-id task-map) 0))
        max-id (or (:bookkeeper/ledger-end-id task-map) Double/POSITIVE_INFINITY)
        {:keys [read-ch shutdown-ch commit-ch]} pipeline
        checkpoint-key (or (:checkpoint/key task-map) task-id)
