      :kafka/pending-messages pending-messages
      :kafka/pending-commits pending-commits}))
 
-(defrecord KafkaReadMessages [max-pending batch-size batch-timeout pending-messages 
+(defrecord KafkaReadMessages [max-pending batch-size batch-timeout pending-messages
                               pending-commits drained? read-ch]
   p-ext/Pipeline
-  (write-batch 
+  (write-batch
     [this event]
     (function/write-batch event))
 

      (swap! pending-commits conj offset))
    (swap! pending-messages dissoc segment-id))

  (retry-segment 
    [_ _ segment-id]
    (when-let [msg (get @pending-messages segment-id)]
      (swap! pending-messages dissoc segment-id)

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

(defn close-read-messages
