              ":"
              (get-in endpoints [:plaintext :port]))))
      {}
-     (k-cluster/all-brokers config))))
+     (k-cluster/all-brokers zk-utils))))
 
 (defn find-brokers [zk-addr]
   (let [results (vals (id->broker zk-addr))]

        commit-interval (or (:kafka/commit-interval task-map) (:kafka/commit-interval defaults))
        client-id "onyx"
        zk-utils (k-admin/make-zk-utils {:servers (:kafka/zookeeper task-map)} false)
        partitions (first (vals (k-partitions/partitions-for zk-utils [topic])))
        n-partitions (count partitions)
        _ (check-num-peers-equals-partitions task-map n-partitions)
        retry-ch (:retry-ch pipeline)
        pending-messages (:pending-messages pipeline)
        pending-commits (:pending-commits pipeline)
        job-id (:onyx.core/job-id event)
        peer-id (:onyx.core/id event)
        task-id (:onyx.core/task-id event)
        done-unsupported? (and (> (count partitions) 1)
                               (not (:kafka/partition task-map)))
        kpartition (if partition
                     (Integer/parseInt (str partition))
                     (log-id @replica job-id peer-id task-id))
        consumer-config {:bootstrap.servers (find-brokers (:kafka/zookeeper task-map))
                         :group.id group-id
                         :enable.auto.commit false

        key-deserializer (byte-array-deserializer)
        value-deserializer (byte-array-deserializer)
        consumer (consumer/make-consumer consumer-config key-deserializer value-deserializer)
        kpartition (if partition
                     (Integer/parseInt (str partition))
                     (log-id @replica job-id peer-id task-id))
        partitions (mapv :partition (metadata/partitions-for consumer topic))
        n-partitions (count partitions)
        done-unsupported? (and (> (count partitions) 1)
                               (not (:kafka/partition task-map)))
        _ (check-num-peers-equals-partitions task-map n-partitions)
        _ (assign-partitions! consumer [{:topic topic :partition kpartition}])
        _ (seek-offset! log consumer group-id topic kpartition task-map)
        offset (cp/next-offset consumer {:topic topic :partition kpartition})
        commit-interval (or (:kafka/commit-interval task-map) (:kafka/commit-interval defaults))
        commit-fut (future (commit-loop log group-id topic kpartition commit-interval pending-commits))]
    {:kafka/commit-fut commit-fut
