 
 (defrecord BookKeeperLogInput
   [log task-id max-pending batch-size batch-timeout pending-messages drained?
-   top-index top-acked-index pending-indexes error read-ch commit-ch shutdown-ch]
+   top-index top-acked-index pending-indexes error retry-ch read-ch commit-ch shutdown-ch]
   p-ext/Pipeline
   (write-batch
     [this event]

  (retry-segment
    [_ event segment-id]
    (when-let [msg (get @pending-messages segment-id)]
      (>!! read-ch (assoc msg :id (random-uuid))))
    (swap! pending-messages dissoc segment-id))

  (pending?


(defn close-read-ledgers-resources
  [{:keys [bookkeeper/producer-ch bookkeeper/commit-ch bookkeeper/read-ch bookkeeper/shutdown-ch] :as event} lifecycle]
  (info "Closing read ledger resources:" (:onyx.core/task event))
  (close! shutdown-ch)
  (close! read-ch)
  ;; Drain the read-ch to unblock the producer
  (while (poll! read-ch))
  (close! commit-ch)
  (let [timeout-ms 15000] 
    (when-not (= producer-ch (second (alts!! [producer-ch (timeout timeout-ms)])))
      (warn (format "Timed out after %s ms waiting for BookKeeper plugin producer thread to finish." timeout-ms))))
  {})

(defn set-starting-offset! [log task-map checkpoint-key start]
