                           top-id
                           top-acked-id
                           pending-indexes
+                          error
                           read-ch
                           commit-ch
                           shutdown-ch)))


  (read-batch
    [_ event]
    (let [pending (count @pending-messages)
          max-segments (min (- max-pending pending) batch-size)
          timeout-ch (timeout batch-timeout)

                          (if (= exit :finished)
                            (>!! read-ch (t/input (random-uuid) :done))))
                        (catch Exception e
                          (reset! error e)
                          (fatal e "BookKeeper plugin: error reading."))))]
    {:bookkeeper/read-ch read-ch
     :bookkeeper/shutdown-ch shutdown-ch
