                     (atom -1)
                     (atom #{})
                     read-ch
+                    retry-ch
                     commit-ch)))
 
 (defn read-datoms [pipeline-data]


  (let [start-tx (:datomic/log-start-tx task-map)
        max-tx (:datomic/log-end-tx task-map)
        {:keys [read-ch shutdown-ch commit-ch]} pipeline
        checkpoint-key (or (:checkpoint/key task-map) task-id)
        _ (set-starting-offset! log task-map checkpoint-key start-tx)
        checkpointed (extensions/read-chunk log :chunk checkpoint-key)

          batch (if (zero? max-segments)
                  (<!! timeout-ch)
                  (->> (range max-segments)
                       (keep (fn [_] (first (alts!! [retry-ch read-ch timeout-ch] :priority true))))))]
      (doseq [m batch]
        (feedback-producer-exception! m)
        (update-top-txes! m top-tx pending-txes)
        (swap! pending-messages assoc (:id m) m))
      (when (completed? batch pending-messages read-ch retry-ch)
        (when-not (:checkpoint/key (:onyx.core/task-map event))
          (>!! commit-ch {:status :complete}))
        (reset! drained? true))
