         pending-messages (atom {})
         pending-commits (atom (sorted-set))
         drained? (atom false)]
-    (->KafkaReadMessages max-pending batch-size batch-timeout 
+    (->KafkaReadMessages max-pending batch-size batch-timeout
                          pending-messages pending-commits drained? ch)))
 
 (defn close-read-messages

                (let [fetched (kc/messages consumer "onyx" topic kpartition head-offset fetch-size)]
                  (Thread/sleep empty-read-back-off)
                  (recur fetched head-offset))
                (let [next-offset (.offset ^int (first ms))]
                  (>!! ch (assoc (t/input (java.util.UUID/randomUUID)
                                          (deserializer-fn (.value ^KafkaMessage (first ms))))
                                 :offset next-offset))
                  (recur (rest ms) (inc next-offset)))))
            (finally

    [_ _ segment-id]
    (get @pending-messages segment-id))

  (drained?
    [_ _]
    @drained?))

