   (validate-task-map! task-map BookKeeperOutput)
   (let [onyx-id (:onyx/tenancy-id peer-opts)
         ledgers-root-path (or (:bookkeeper/zookeeper-ledgers-root-path task-map)
-                              (:onyx.bookkeeper/zk-ledgers-root-path defaults))
+                              (:onyx.bookkeeper/zk-ledgers-root-path default-vals))
         zookeeper-addr (:bookkeeper/zookeeper-addr task-map)
         ;; FIXME, parameterize these in the task-map
         zookeeper-timeout 60000

    @drained?))

(defn read-ledgers [{:keys [onyx.core/task-map onyx.core/log onyx.core/task-id] :as pipeline-data}]
  (let [max-pending (or (:onyx/max-pending task-map) (:onyx/max-pending defaults))
        batch-size (:onyx/batch-size task-map)
        batch-timeout (or (:onyx/batch-timeout task-map) (:onyx/batch-timeout defaults))
        read-ch (chan (or (:bookkeeper/read-buffer task-map) 1000))
        retry-ch (chan (* 2 max-pending))
        error (atom nil)

  [{:keys [onyx.core/task-map onyx.core/log onyx.core/peer-opts onyx.core/task-id onyx.core/job-id] :as event} lifecycle]
  (let [onyx-id (:onyx/tenancy-id peer-opts)
        ledgers-root-path (or (:bookkeeper/zookeeper-ledgers-root-path lifecycle)
                              (:onyx.bookkeeper/zk-ledgers-root-path default-vals))
        zookeeper-addr (:bookkeeper/zookeeper-addr lifecycle)
        ;; FIXME, parameterize these in the lifecycle
        zookeeper-timeout 60000
