     [_ _ message-id]
     (when-let [msg (get @pending-messages message-id)]
       (swap! pending-messages dissoc message-id)
-      (>!! read-ch msg)))
+      (>!! read-ch (t/input (java.util.UUID/randomUUID)
+                            (:message msg)))))
 
   (pending?
     [_ _ message-id]

  (if (:kafka/force-reset? task-map)
    (read-from-bound consumer topic kpartition task-map)
    (if-let [x (kzk/committed-offset m group-id topic kpartition)]
      x
      (read-from-bound consumer topic kpartition task-map))))

(defn highest-offset-to-commit [offsets]

                  (Thread/sleep empty-read-back-off)
                  (recur fetched head-offset))
                (let [next-offset (.offset ^int (first ms))]
                  (>!! ch (assoc (t/input (java.util.UUID/randomUUID)
                                          (deserializer-fn (.value ^KafkaMessage (first ms))))
                                 :offset next-offset))
                  (recur (rest ms) (inc next-offset)))))
            (finally
             (future-cancel commit-fut))))
