   (ack-segment [_ _ segment-id]
     (let [chunk-index (:chunk-index (@pending-messages segment-id))]
       (swap! pending-chunk-indices disj chunk-index)
-      (let [new-top-acked (highest-acked-chunk @top-acked-chunk-index @top-chunk-index @pending-chunk-indices)
-            updated-content {:chunk-index new-top-acked :status :incomplete}]
-        (extensions/force-write-chunk log :chunk updated-content task-id)
+      (let [new-top-acked (highest-acked-chunk @top-acked-chunk-index @top-chunk-index @pending-chunk-indices)]
+        (>!! commit-ch {:chunk-index new-top-acked :status :incomplete})
         (reset! top-acked-chunk-index new-top-acked))
       (swap! pending-messages dissoc segment-id)))
 

        max-pending (or (:onyx/max-pending catalog-entry) (:onyx/max-pending defaults))
        batch-size (:onyx/batch-size catalog-entry)
        batch-timeout (or (:onyx/batch-timeout catalog-entry) (:onyx/batch-timeout defaults))
        read-ch (chan (or (:datomic/read-buffer catalog-entry) 1000))] 
    (->DatomicInput (:onyx.core/log pipeline-data)
                    (:onyx.core/task-id pipeline-data)
                    max-pending batch-size batch-timeout 

                    (atom -1)
                    (atom -1)
                    (atom #{})
                    read-ch
                    commit-ch)))

(defn read-datoms [pipeline-data]
  (shared-input-builder pipeline-data))
