                                            (seq (drop datoms-per-segment datoms)))))))
                             (>!! ch (t/input (random-uuid) :done))
                             (catch Exception e
-                              (fatal e))))]
+                              ;; feedback exception to read-batch
+                              (>!! ch e))))]
 
         {:datomic/read-ch ch
          :datomic/commit-ch (:commit-ch pipeline)

                  (->> (range max-segments)
                       (keep (fn [_] (first (alts!! [read-ch timeout-ch] :priority true))))))]
      (doseq [m batch]
        (let [message (:message m)]
          (when-not (= message :done)
            (swap! top-tx max (:t message))
            (swap! pending-txes conj (:t message))))
        (swap! pending-messages assoc (:id m) m))
      (when (and (all-done? (vals @pending-messages))
                 (all-done? batch)
                 (zero? (count (.buf read-ch)))
                 (or (not (empty? @pending-messages))
                     (not (empty? batch))))
        (when-not (:checkpoint/key (:onyx.core/task-map event))
          (>!! commit-ch {:status :complete}))
        (reset! drained? true))

  (empty? (remove #(= :done (:message %))
                  messages)))

(defn completed? [batch pending-messages read-ch]
  (and (all-done? (vals @pending-messages))
       (all-done? batch)
       (zero? (count (.buf read-ch)))
       (or (not (empty? @pending-messages))
           (not (empty? batch)))))

(defn feedback-producer-exception! [e]
  (when (instance? java.lang.Throwable e)
    (throw e)))

(defn update-chunk-indices! [m top-chunk-index pending-chunk-indices]
  (when-let [chunk-index (:chunk-index m)]
    (swap! top-chunk-index max chunk-index)
    (swap! pending-chunk-indices conj chunk-index)))

(defrecord DatomicInput [log task-id max-pending batch-size batch-timeout
                         pending-messages drained?
                         top-chunk-index top-acked-chunk-index pending-chunk-indices
