         client-id "onyx"
         m {"zookeeper.connect" (:kafka/zookeeper task-map)}
         partitions (kzk/partitions m topic)
-        ch (:read-ch pipeline) 
+        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))]


(defrecord KafkaWriteMessages [config topic producer serializer-fn]
  p-ext/Pipeline
  (read-batch 
    [_ event]
    (function/read-batch event))

  (write-batch 
    [_ {:keys [onyx.core/results]}]
    (let [messages (mapcat :leaves (:tree results))]
      (doseq [m (map :message messages)]
        (kp/send-message producer (kp/message topic (serializer-fn m)))))
    {})

  (seal-resource 
    [_ {:keys [onyx.core/results]}]
    (kp/send-message producer (kp/message topic (serializer-fn :done)))))


        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
