   (checking
    "Functions that throw exceptions pass the exception object back"
    (times 30)
-   [input (gen/vector (gen/hash-map :message (gen/map gen/any gen/any)))]
+   [input (gen/vector (gen-segment))]
    (let [f (fn [segment] (throw (ex-info "exception" {:val 42})))
          event {:onyx.core/batch input}
          rets (t/apply-fn f false event)

  (checking
   "Multiple segments can be returned, attached to their root"
   (times 30)
   [input (gen/hash-map :message (gen/map gen/any gen/any))
    output (gen/vector (gen/map gen/any gen/any))]
   (let [f (fn [segment] output)
         event {:onyx.core/batch [input]}

     (is (= input (map :root tree)))
     (is (every? #(= clojure.lang.ExceptionInfo (type %)) messages))
     (is (every? #(= {:val 42} (ex-data (:exception %))) (map ex-data messages))))))

(deftest bulk-functions
  (checking
   "Bulk functions call the function, but always return their inputs"
   (times 30)
   [input (gen/not-empty (gen/vector (gen-segment)))
    output gen/any]
   (let [called? (atom false)
         f (fn [segment] (reset! called? true) output)
         event {:onyx.core/batch input}
         rets (t/apply-fn f true event)
         tree (:tree (t/persistent-results! (:onyx.core/results rets)))]
     (is (= input (map :root tree)))
     (is (= (map :message input) (map :message (mapcat :leaves tree))))
     (is @called?))))
