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

            [onyx.peer.pipeline-extensions :as p-ext]
            [onyx.peer.function :as function]
            [onyx.types :as t]
            [onyx.static.default-vals :refer [default-vals]]
	    [clojure.core.async.impl.protocols :refer [closed?]]
            [onyx.log.zookeeper :as log-zk]
            [onyx.log.curator :as zk]
