         commit-interval (or (:kafka/commit-interval task-map) (:kafka/commit-interval defaults))
         client-id "onyx"
         m {"zookeeper.connect" (:kafka/zookeeper task-map)}
-        partitions (:partitions pipeline)
+        partitions (kzk/partitions m (:kafka/topic task-map))
         n-partitions (count partitions)
         _ (check-num-peers-equals-partitions task-map n-partitions)
         ch (:read-ch pipeline)

        task-id (:onyx.core/task-id event)
        reader-fut (future (reader-loop m client-id group-id topic partition partitions task-map 
                                        (:onyx.core/replica event) job-id peer-id task-id
                                        ch pending-commits))]
    {:kafka/read-ch ch
     :kafka/reader-future reader-fut
     :kafka/pending-messages pending-messages
     :kafka/pending-commits pending-commits}))

(defn all-done? [messages]
  (empty? (remove #(= :done (:message %))
                  messages)))

(defrecord KafkaReadMessages [max-pending batch-size batch-timeout partitions done-unsupported? 
                              pending-messages pending-commits drained? read-ch]
  p-ext/Pipeline
  (write-batch
    [this event]

        (swap! pending-messages assoc (:id m) m))
      (when (and (all-done? (vals @pending-messages))
                 (all-done? batch)
                 (zero? (count (.buf read-ch)))
                 (or (not (empty? @pending-messages))
                     (not (empty? batch))))
        (if (:kafka/done-unsupported? event)
          (throw (UnsupportedOperationException. ":done is not supported for auto assigned kafka partitions. 
                                                 (:kafka/partition must be supplied)"))
          (reset! drained? true)))
