             (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
+        {:keys [read-ch shutdown-ch retry-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))

        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)

                 (all-done? batch)
                 (or (not (empty? @pending-messages))
                     (not (empty? batch)))
                 (zero? (count (.buf retry-ch)))
                 (zero? (count (.buf read-ch))))
        (when-not (:checkpoint/key (:onyx.core/task-map event))
          (>!! commit-ch {:status :complete}))
