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

                (let [message ^KafkaMessage (first ms)
                      next-offset ^int (.offset message)
                      dm (deserializer-fn (.value message))]
                  (>!! ch (assoc (t/input (java.util.UUID/randomUUID)
                                          dm)
                                 :offset next-offset))
                  (recur (rest ms) (inc next-offset)))))
            (finally

            [zookeeper :as zk]
            [onyx.static.default-vals :refer [arg-or-default]]
            [onyx.types :as t]
            [onyx.static.uuid :refer [random-uuid]]
            [onyx.peer.pipeline-extensions :as p-ext]
            [onyx.peer.function :as function]
            [onyx.peer.operation :refer [kw->fn]]
