 
 (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}]

                          (let [metadata (get-metadata metadata-producers conf)]
	                          (dosync 
                              (alter brokers-metadata (fn [x] metadata))
                              (alter producer-ref (fn [x] (update-producers @metadata @x)))
                              )))
                           
        metadata-fixdelay 

                           (fn [x] 
                             (if-let [producer (get @producers-ref k)]
                               x ; return map the producer already exists
                               (let [[partition broker] (select-broker topic partition brokers-metadata)]
	                               (if-let [producer (find-hashed-connection broker k @producers-ref (get conf :producer-connections-max 4))]
	                                 (assoc x k producer) ; found a producer based on hash and broker for the same partition
	                                 (assoc x k (delay ;else create a new producer
		                                            (let [producer-buffer (create-producer-buffer connector topic partition producer-error-ch broker conf)]
			                                             (add-remove-on-error-listener! producers-ref topic k producer-buffer brokers-metadata)
		                                               producer-buffer)))))))))
            k)))))
	     

