     (returning nil (.close rr'))
     (returning rr' (close-rr rr))))
 
+(defn ^:private rr-tuple
+  [^RecordReader rr]
+  [(.getCurrentKey rr) (.getCurrentValue rr)])
+
 (defn records-seqable
   [conf f klass & paths]
   (let [job (doto-let [job (mr/job conf)]
               (doseq [path paths :let [path (fs/path path)]]
                 (FileInputFormat/addInputPath job path)))
-        conf (.getConfiguration job)
+        conf (conf/ig job)
         ^InputFormat ifi (ReflectionUtils/newInstance klass conf)
         splits (seq (.getSplits ifi job))
         closer (agent ::initial)]
     (reify
-      Closeable (close [_] (send closer close-rr))
+      Closeable
+      (close [_] (send closer close-rr))
+
       Seqable
       (seq [_]
         ((fn step [^RecordReader rr splits]

                           (.initialize split tac))]
                  (returning (step rr splits)
                    (send closer update-rr rr)))))))
         nil splits)))))

(ns parkour.inspect.mapreduce
  {:private true}
  (:require [clojure.core.protocols :as ccp]
            [parkour (mapreduce :as mr) (fs :as fs) (conf :as conf)]
            [parkour.util :refer [ignore-errors doto-let returning]])
  (:import [java.io Closeable]
           [clojure.lang Seqable]
