    (:added datom)])
 
 (defn close-read-log-resources
-  [{:keys [datomic/producer-ch datomic/commit-ch datomic/read-ch datomic/shutdown-ch] :as event} lifecycle]
+  [{: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)

          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}))

                          (>!! read-ch e))))]

    {:datomic/read-ch read-ch
     :datomic/retry-ch retry-ch
     :datomic/shutdown-ch shutdown-ch
     :datomic/commit-ch commit-ch
     :datomic/producer-ch producer-ch
