                         (let [exit (loop [last-acked (inc (:largest checkpointed))]
                                      (read-fn client ledger-id digest password read-ch deserializer-fn backoff-period last-acked max-id)
                                      :finished)]
-                          (if-not (= exit :shutdown)
+                          (if (= exit :finished)
                             (>!! read-ch (t/input (random-uuid) :done))))
                         (catch Exception e
-                          (fatal e))))]
+                          (>!! read-ch (t/input (random-uuid) :crash))
+                          (fatal "BookKeeper plugin: error reading." e))))]
     {:bookkeeper/read-ch read-ch
      :bookkeeper/shutdown-ch shutdown-ch
      :bookkeeper/commit-ch commit-ch


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

(defn set-starting-offset! [log task-map checkpoint-key start]

                           (keep (fn [_] (first (alts!! [read-ch timeout-ch] :priority true))))))]
      (doseq [m batch]
        (let [message (:message m)]
          (when (= message :crash)
            (throw (Exception. "Plugin crashed. Crash read.")))

          (when-not (= message :done)
            (swap! top-index max (:entry-id message))
            (swap! pending-indexes conj (:entry-id message))))
