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

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

                       (map (fn [_]
                              (let [result (first (alts!! [read-ch timeout-ch] :priority true))]
                                (if (= (:message result) :done)
                                  (t/input (random-uuid) :done)
                                  result))))
                       (filter :message)))]
      (doseq [m batch]
