           segments (->> (flush-swap! retry 
                                      #(take max-segments %)
                                      #(subvec % (min max-segments (count %))))
-                        (map (fn [m] (t/input (uuid/v1) m))))
+                        (map (fn [m] (t/input (random-uuid) m))))
           batch (loop [n (count segments) 
                        sgs segments]
                   (if (= n max-segments)
                     sgs
                     (recur (inc n)
-                           (conj sgs (t/input (uuid/v1) (new-segment-fn n))))))]
+                           (conj sgs (t/input (random-uuid) (new-segment-fn n))))))]
       (doseq [m batch] 
         (swap! pending-messages assoc (:id m) (:message m)))
       {:onyx.core/batch batch}))

(defn new-segment-small [batch-index]
  {:n batch-index :data hundred-bytes})

(defn new-grouping-segment [batch-index]
  {:id (uuid/v1)
   :event-time (java.util.Date.)
   :group-key (rand-int 10000)
   :value (rand-int 500)})

            [onyx.peer.function :as function]
            [onyx.static.default-vals :refer [defaults]]
            [onyx.types :as t]
            [taoensso.timbre :refer [info warn trace fatal] :as timbre]
            [onyx.peer.pipeline-extensions :as p-ext]))

