                           (reset! error e)
                           (fatal e "BookKeeper plugin: error reading."))))]
     {:bookkeeper/read-ch read-ch
+     :bookkeeper/retry-ch retry-ch
      :bookkeeper/shutdown-ch shutdown-ch
      :bookkeeper/commit-ch commit-ch
      :bookkeeper/producer-ch producer-ch

            (default-value :bookkeeper/ledger-end-id Double/POSITIVE_INFINITY)
            (default-value :bookkeeper/no-recovery-empty-read-back-off 500)
            (default-value :checkpoint/key task-id))
        {:keys [read-ch shutdown-ch commit-ch error]} pipeline
        ;; decrement because we are going to store this as a checkpoint and then inc after recover
        checkpoint-key (:checkpoint/key defaulted-task-map)
        _ (set-starting-offset! log task-map checkpoint-key (dec ledger-start-id))

;; read BookKeeper log plugin

(defn close-read-ledgers-resources
  [{:keys [bookkeeper/producer-ch bookkeeper/commit-ch bookkeeper/read-ch bookkeeper/retry-ch bookkeeper/shutdown-ch] :as event} lifecycle]
  (info "Closing read ledger resources:" (:onyx.core/task event))
  (close! shutdown-ch)
  (close! read-ch)
  (close! retry-ch)
  ;; Drain the read-ch to unblock the producer
  (while (poll! read-ch))
  (while (poll! retry-ch))
  (close! commit-ch)
  (let [timeout-ms 15000] 
    (when-not (= producer-ch (second (alts!! [producer-ch (timeout timeout-ms)])))
