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

  (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)))]
   (let [f (fn [segment] (throw (ex-info "exception" {:val 42})))
         event {:onyx.core/batch input}
         rets (t/apply-fn f false event)

  (checking
   "Functions can be parameterized via the event map"
   (times 30)
   [input (gen/vector (gen-segment))
    params (gen/vector gen/any)]
   (let [f (fn [& args] {:result (or (butlast args) [])})
         event {:onyx.core/batch input :onyx.core/params params}
