                                (into {} (filter (complement nil?)
                                             (map (fn [[k producer-buffer]]
 								                                                    (let [[topic partition] (clojure.string/split k #"\:")
-								                                                          host (:host producer-buffer)]
-								                                                      (if (some #(= (:host %) host) (get metadata topic))
+								                                                          host (smart-get producer-buffer :host)]
+								                                                      (if (some #(= (:host %) host) (smart-get metadata topic))
 								                                                        [k producer-buffer]
 								                                                        (do
-                                                                          (prn "closing producer " host " topic " topic)
+                                                                          (prn "closing producer " host " topic " topic " prodbuffer " producer-buffer)
 								                                                          (close-producer-buffer! producer-buffer)
 								                                                          nil))))
                                               m))))

        )))))
     
  
(defn find-hashed-connection [k producers upper-limit]
  "Finds the value with a key that when hashed and moded against upper-limit is the same for k
   this is an optimization for hashed partitioning where the original keys are stored in the producers-ref
   i.e. topic:partition but the connections are created on (mod (hash k) upper-limit)"
  (let [hashed-k (hash k)
	     [_ producer] (first 
                       (filter (fn [[producer-k v]]
                         (= (mod (hash producer-k) upper-limit) (mod hashed-k upper-limit))) producers))]
      producer))
	             
             
(defn select-producer-buffer! [connector topic partition {:keys [producers-ref brokers-metadata producer-error-ch conf] :as state}]
  "Select a producer or create one,


(defn- get-partition-count [topic brokers-metadata]
  (count (get brokers-metadata topic)))


(defn smart-get [x k]
  (if (or (delay? x) (future? x))
    (get @x k)
    (get x k)))


(defn select-rr-partition! [topic {:keys [topic-partition-ref brokers-metadata] :as state}]
