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

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

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


        client-id "onyx"
        m {"zookeeper.connect" (:kafka/zookeeper task-map)}
        partitions (kzk/partitions m topic)
        ch (:read-ch pipeline)
        pending-messages (:pending-messages pipeline)
        pending-commits (:pending-commits pipeline)
        reader-fut (future (reader-loop m client-id group-id topic partitions kpartition task-map ch pending-commits))]
