         batch-timeout (or (:onyx/batch-timeout catalog-entry) (:onyx/batch-timeout defaults))
         read-ch (chan (or (:datomic/read-buffer catalog-entry) 1000))
         commit-ch (chan (sliding-buffer 1))
+        shutdown-ch (chan 1)
         top-tx (atom -1)
         top-acked-tx (atom -1)
         pending-txes (atom #{})]

   (:added datom)])

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

(defn set-starting-offset! [log task-map checkpoint-key start-tx]
  (if (:checkpoint/force-reset? task-map)

(defrecord DatomicLogInput
  [log task-id max-pending batch-size batch-timeout pending-messages drained?
   top-tx top-acked-tx pending-txes
   read-ch commit-ch shutdown-ch]
  p-ext/Pipeline
  (write-batch
    [this event]
