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

         (let [klass (class split)]
           (.writeInt out i)
           (Text/writeString out (.getName klass))
           (doto (-> conf SerializationFactory. (.getSerializer klass))
             (.open out)
             (.serialize 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")
