             [franzy.clients.producer.types :refer [make-producer-record]]
             [franzy.clients.consumer.protocols :as proto]
             [franzy.clients.consumer.client :as consumer]
+            [franzy.common.metadata.protocols :as metadata]
             [franzy.clients.consumer.protocols :refer [assign-partitions! commit-offsets-sync!
                                                        poll! seek-to-offset!] :as cp]
             [taoensso.timbre :as log :refer [fatal info]]

;; kafka operations

(defn id->broker [zk-addr]
  (let [config (k-admin/make-zk-utils {:servers zk-addr} false)]
    (reduce
     (fn [result {:keys [id endpoints]}]
       (assoc

             ":"
             (get-in endpoints [:plaintext :port]))))
     {}
     (k-cluster/all-brokers zk-utils))))

(defn find-brokers [zk-addr]
  (let [results (vals (id->broker zk-addr))]
