   ^Combine$CombineFn
   ([reducef extractf combinef initf output-coder acc-coder]
    (proxy [Combine$CombineFn ICombineFn] []
-     (createAccumulator [] (initf))
-     (addInput [acc elt] (reducef acc elt))
-     (mergeAccumulators [accs] (apply combinef accs))
-     (extractOutput [acc] (extractf acc))
+     (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)

  [f]
  (proxy [SerializableFunction] []
    (apply [input]
      (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)]
