   "
   (info "Connecting to redis using " (get conf :redis-conf {:heart-beat-freq 10}))
   (let [
+        offset-producers (ref {})
         metrics (create-metrics)
         msg-ch (chan 100)
         msg-buff (buffered-chan msg-ch 1000 1000 1000)

         (throw (RuntimeException. (str "Error reading offsets from " offset-producer " for topic " topic " error: " v))))
       (throw (RuntimeException. (str "Timeout while reading offsets from " offset-producer " for topic " topic))))))

(defn get-broker-offsets [metadata topics conf]
  "Builds the datastructure {broker {topic [{:offset o :partition p} ...] }}"
   (apply merge-with merge
     (for [topic topics] 

	
			      ))))

(defn consume [conn bootstrap-brokers group-conn msg-ch topics conf]
  "Entry point for topic consumption,
   The cluster metadata is requested from the bootstrap-brokers, the topic offsets are sorted per broker.
   For each broker a producer is created that will control the sending and reading from the broker,
   then consume-producers is called in the background that will reconnect if needed,
   the method returns with {:msg-ch and :shutdown (fn []) }, shutdown should be called to stop all consumption for this topic"
  (if-let [metadata (get-metadata bootstrap-brokers {})]
    (let[broker-offsets (doall (get-broker-offsets conn metadata topics conf))
         producers (doall (create-producers broker-offsets conf))
         t (future (try
                     (consume-producers! conn bootstrap-brokers group-conn producers topics broker-offsets msg-ch conf)
                     (catch Exception e (error e e))))]
      {:msg-ch msg-ch :shutdown (fn [] (future-cancel t))}
      )
