          (let [klass (class split)]
            (.writeInt out i)
            (Text/writeString out (.getName klass))
-           (doto (-> conf SerializationFactory. (.getSerializer klass))
+           (doto (serializer conf klass)
              (.open out)
              (.serialize split))))
 

       (readSplit [_ in]
         (let [i (.readInt in)
               klass (->> in Text/readString (.getClassByName conf))
               split (-> (doto (-> conf SerializationFactory.
                                   (.getDeserializer klass))
                           (.open in))
                         (.deserialize (w/new-instance conf klass)))]
           (input-split* conf i split)))

(defn ^:private get-subconfs
  [job] (edn/read-string (conf/get job confs-key)))

(defn ^:private serializer
  {:tag `Serializer}
  [conf klass] (-> conf SerializationFactory. (.getSerializer klass)))

(defn ^:private deserializer
  {:tag `Deserializer}
  [conf klass] (-> conf SerializationFactory. (.getDeserializer klass)))

(defn ^:private input-split*
  ([] (input-split* nil nil nil))
  ([conf] (input-split* conf nil nil))
