   [f]
   (proxy [SerializableFunction] []
     (apply [input]
-      (f input))))
+      (safe-exec (f input)))))
 
 (definterface ICombineFn
   (getReduceFn [])

      (.registerCoder clojure.lang.Keyword (make-transit-coder)))
    pipeline))


(defn write-text-file
  ([to options ^PCollection pcoll]
   (let [opts (assoc options :label :write-text-file)]

  ^Combine$CombineFn
  ([reducef extractf combinef initf output-coder acc-coder]
   (proxy [Combine$CombineFn ICombineFn] []
     (createAccumulator [] (safe-exec (initf)))
     (addInput [acc elt] (safe-exec (reducef acc elt)))
     (mergeAccumulators [accs] (safe-exec (apply combinef accs)))
     (extractOutput [acc] (safe-exec (extractf acc)))
     (getDefaultOutputCoder [_ _] output-coder)
     (getAccumulatorCoder [_ _] acc-coder)
     (getReduceFn [] reducef)
