                           (if-not (= exit :shutdown)
                             (>!! read-ch (t/input (random-uuid) :done))))
                         (catch Exception e
-                          (fatal e))))]
+                          ;; feedback exception to read-batch
+                          (>!! read-ch e))))]
 
     {:datomic/read-ch read-ch
      :datomic/shutdown-ch shutdown-ch

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

          batch (->> (range max-segments)
                     (keep (fn [_] (first (alts!! [read-ch timeout-ch] :priority true)))))]
      (doseq [m batch]
        (feedback-producer-exception! m)
        (update-chunk-indices! m top-chunk-index pending-chunk-indices)
        (swap! pending-messages assoc (:id m) m))
    (when (completed? batch pending-messages read-ch) 
      (>!! commit-ch {:status :complete})
      (reset! drained? true))
    {:onyx.core/batch batch}))
