   (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

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

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

        {:datomic/read-ch ch
         :datomic/commit-ch (:commit-ch pipeline)
