     [_ _ 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]

    (loop []
      (Thread/sleep commit-interval)
      (when-let [offset (highest-offset-to-commit @pending-commits)]
        (log/info "Writing offset to ZooKeeper: " offset)
        (kzk/set-offset! m group-id topic kpartition offset)
        (swap! pending-commits (fn [coll] (remove (fn [k] (<= k offset)) coll))))
      (recur))

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