   (empty? (remove #(= :done (:message %))
                   messages)))
 
-(defn completed? [batch pending-messages read-ch]
+(defn completed? [batch pending-messages read-ch retry-ch]
   (and (all-done? (vals @pending-messages))
        (all-done? batch)
        (zero? (count (.buf read-ch)))
+       (zero? (count (.buf retry-ch)))
        (or (not (empty? @pending-messages))
            (not (empty? batch)))))
 

          max-segments (min (- max-pending pending) batch-size)
          timeout-ch (timeout batch-timeout)
          batch (->> (range max-segments)
                     (keep (fn [_] (first (alts!! [read-ch timeout-ch] :priority true)))))]
      (doseq [m batch]
        (feedback-producer-exception! m)
        (update-chunk-indices! m top-chunk-index pending-chunk-indices)
        (swap! pending-messages assoc (:id m) m))
      (when (completed? batch pending-messages read-ch)
        (>!! commit-ch {:status :complete})
        (reset! drained? true))
      {:onyx.core/batch batch}))

   (:added datom)])

(defn close-read-log-resources
  [{:keys [datomic/producer-ch datomic/commit-ch datomic/read-ch datomic/retry-ch datomic/shutdown-ch] :as event} lifecycle]
  (close! read-ch)
  (close! retry-ch)
  (while (poll! read-ch))
  (while (poll! retry-ch))
  (close! commit-ch)
  (close! shutdown-ch)
  (<!! producer-ch)
