 
 (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]
+   top-index top-acked-index pending-indexes error read-ch commit-ch shutdown-ch]
   p-ext/Pipeline
   (write-batch
     [this event]

      (doseq [m batch]
        (let [message (:message m)]
          (info "Read message in batch" message)
          (when (instance? java.lang.Throwable message)
            (throw message))

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

        batch-size (:onyx/batch-size task-map)
        batch-timeout (or (:onyx/batch-timeout task-map) (:onyx/batch-timeout defaults))
        read-ch (chan (or (:bookkeeper/read-buffer task-map) 1000))
        error (atom nil)
        commit-ch (chan (sliding-buffer 1))
        shutdown-ch (chan 1)
        top-id (atom -1)
