   (:metadata-error-ch connector))
 
 
-(defn create-connector [bootstrap-brokers {:keys [acks batch-fail-message-over-limit batch-byte-limit blacklisted-expire producer-retry-strategy] :or {blacklisted-expire 10000 acks 0 batch-fail-message-over-limit true batch-byte-limit 10485760 producer-retry-strategy :default} :as conf}]
+(defn create-connector [bootstrap-brokers {:keys [acks batch-fail-message-over-limit batch-byte-limit blacklisted-expire producer-retry-strategy topic-auto-create] :or {blacklisted-expire 10000 acks 0 batch-fail-message-over-limit true batch-byte-limit 10485760 producer-retry-strategy :default topic-auto-create true} :as conf}]
   (let [
         ^ScheduledExecutorService scheduled-service (Executors/newSingleThreadScheduledExecutor)
         metadata-producers-ref (ref (filter (complement nil?) (map #(delay (metadata-request-producer (:host %) (:port %) conf)) bootstrap-brokers)))

                   (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)]
                       (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)))
    (if (-> state :conf :topic-auto-create)
      0
      (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"
  (if-let [brokers (get brokers-metadata topic)]
    (rand-nth brokers)
    (->> brokers-metadata keys (rand-nth) (get brokers-metadata) rand-nth)))

  
(defn select-broker
  "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"
  [topic partition brokers-metadata & {:keys [topic-auto-create]}]
  (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))
