                           (if (= exit :finished)
                             (>!! read-ch (t/input (random-uuid) :done))))
                         (catch Exception e
-                          (>!! read-ch (t/input (random-uuid) e))
+                          (reset! error e)
                           (fatal e "BookKeeper plugin: error reading."))))]
     {:bookkeeper/read-ch read-ch
      :bookkeeper/shutdown-ch shutdown-ch


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

            (default-value :bookkeeper/ledger-end-id Double/POSITIVE_INFINITY)
            (default-value :bookkeeper/no-recovery-empty-read-back-off 500)
            (default-value :checkpoint/key task-id))
        {:keys [read-ch shutdown-ch commit-ch error]} pipeline
        ;; decrement because we are going to store this as a checkpoint and then inc after recover
        checkpoint-key (:checkpoint/key defaulted-task-map)
        _ (set-starting-offset! log task-map checkpoint-key (dec ledger-start-id))
