 (defn- send-msg-retry [{:keys [state] :as connector} {:keys [topic] :as msg}]
   "Try sending the message to any of the producers till all of the producers have been trieds"
    (do-metadata-update connector)
-   (let [partitions (-> state :brokers-metadata deref (get topic))]
-      (if (or (empty? partitions) (= partitions 0))
-        (throw (RuntimeException. (str "No topics available for topic " topic " partitions " partitions))))
-      
-      (loop [partitions-1 partitions i 0] 
+   (let [partitions  (let [partitions(-> state :brokers-metadata deref (get topic)) ]
+                       (if (or (empty? partitions) (= partitions 0))
+                         (if (-> state :conf :topic-auto-create)
+                           [0]
+                           (throw (RuntimeException. (str "No topics available for topic " topic " partitions " partitions))))
+                         partitions))]
+
+     (prn "using partitions " partitions)
+      (loop [partitions-1 partitions i 0]
         (if-let [partition (first partitions-1)]
           (let [sent (try
 						            (let [producer-buffer (exception-if-blacklisted (select-producer-buffer! connector topic i state) @(:blacklisted-producers-ref state))]

               :topic-partition-ref (ref {})
               :producer-error-ch producer-error-ch
               :blacklisted-producers-ref blacklisted-producers-ref
               :conf (assoc conf :batch-fail-message-over-limit batch-fail-message-over-limit :batch-byte-limit batch-byte-limit)}
        ;go through each producer, if it does not exist in the metadata, the producer is closed, otherwise we keep it.
        update-producers (fn [metadata m] 
                               (into {} (filter (complement nil?)

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