         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 max-pending)
         error (atom nil)
         commit-ch (chan (sliding-buffer 1))
         shutdown-ch (chan 1)

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

            [onyx.static.uuid :refer [random-uuid]]
            [onyx.peer.operation :refer [kw->fn]]
            [onyx.types :refer [dec-count! inc-count!]]
            [taoensso.timbre :refer [info warn error trace debug fatal]])
  (:import [org.apache.zookeeper KeeperException$BadVersionException KeeperException$ConnectionLossException]
           [org.apache.bookkeeper.client 
            LedgerHandle LedgerEntry BookKeeper BKException$Code BKException$ZKException BookKeeper$DigestType AsyncCallback$AddCallback]))
