                    (info "adding producer to ref:  topic " topic " partition " partition)
                    (if-let [producer (-> producers-ref deref (get topic) (get partition))] ;use get get for speed
                      x ; return map the producer already exists
-                     (let [[partition broker] (select-broker topic partition brokers-metadata)]
+                     (let [[partition broker] (select-broker topic partition brokers-metadata :topic-auto-create (:topic-auto-create conf))]
                        (if-let [producer (find-hashed-connection broker (str topic ":" partition) @producers-ref (get conf :producer-connections-max 2))]
                          (assoc-in x [topic partition] producer) ; found a producer based on hash and broker for the same partition
                          (assoc-in x [topic partition] (delay ;else create a new producer

		           (commute topic-partition-ref (fn [x] 
		                                          (assoc x topic (AtomicLong. 0)))))
		     (select-rr-partition! topic state)))
    (do 
      (error "No topic found " topic " in state map " topic-partition-ref)
      (error "broker-meta " brokers-metadata)
      (throw (RuntimeException. 
             (str "The topic " topic " does not exist, please create it first. See http://kafka.apache.org/documentation.html")))))))

(defn get-broker-from-metadata [topic partition brokers-metadata]
  (-> brokers-metadata (get topic) (get partition)))

(defn random-select-broker [topic brokers-metadata]
  "Select a broker over all the registered topics randomly"
  (-> (map (fn [[k v]] [k v]) brokers-metadata) rand-nth rest rand-nth))

  
(defn select-broker [topic partition brokers-metadata]
  "Try to find the broker in the brokers-metadata by topic and partition,
    if no broker can be found, a random select is done, if still
    no broker is found nil is returned"
  (if-let [broker (get-broker-from-metadata topic partition @brokers-metadata)]
    [partition broker]
    [0 (random-select-broker topic brokers-metadata)]))

(defn send-to-buffer [{:keys [ch-source]} msg]
  (>!! ch-source msg))

                                  ))))))))
     
  
(defn find-hashed-connection
  "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)"
  [broker k producers upper-limit]
  (let [hashed-k (hash k)
	     [_ producer] (first 
                       (filter (fn [[producer-k {:keys [host port]}]]
