    (:added datom)])
 
 (defn close-read-log-resources
-  [{:keys [datomic/producer-ch datomic/commit-ch datomic/read-ch] :as event} lifecycle]
+  [{:keys [datomic/producer-ch datomic/commit-ch datomic/read-ch datomic/shutdown-ch] :as event} lifecycle]
   (close! read-ch)
   (close! commit-ch)
-  (close! producer-ch))
+  (close! shutdown-ch)
+  (<!! producer-ch)
+  {})
 
 (defn set-starting-offset! [log task-map checkpoint-key start-tx]
   (if (:checkpoint/force-reset? task-map)

                       top-acked-tx
                       pending-txes
                       read-ch
                       commit-ch)))

(def read-log-calls
  {:lifecycle/before-task-start inject-read-log-resources

(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]
