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

           [org.apache.hadoop.conf Configuration]
           [org.apache.hadoop.io Text]
           [org.apache.hadoop.io.serializer SerializationFactory]
           [org.apache.hadoop.io.serializer Deserializer Serializer]
           [org.apache.hadoop.mapreduce InputFormat InputSplit Job]
           [org.apache.hadoop.mapreduce Mapper Mapper$Context]
           [parkour.hadoop.interfaces IInputFormat IInputSplit]
           [parkour.hadoop.input
             MultiplexInputFormat MultiplexInputSplit MultiplexRecordReader]))

(def ^:private ^:const confs-key
  "parkour.multiplex.confs")
