 	
 			      ))))
 
-(defn consume [bootstrap-brokers group-conn msg-ch topics conf]
+(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 metadata topics conf))
+    (let[broker-offsets (doall (get-broker-offsets conn metadata topics conf))
          producers (doall (create-producers broker-offsets conf))
          t (future (try
-                     (consume-producers! bootstrap-brokers group-conn producers topics broker-offsets msg-ch conf)
+                     (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))}
       )

			        (for [[broker v] by-broker]
			          ;here we have data {{:host "localhost", :port 1} [[0 {:host "localhost", :port 1}] [1 {:host "localhost", :port 1}]], {:host "abc", :port 1} [[2 {:host "abc", :port 1}]]}
			          ;doing map first v gives the partitions for a broker
			          (let [offset-producer (create-offset-producer broker conf)
			                offsets-response (get-offsets offset-producer topic (map first v))]
			            (shutdown offset-producer)
			            [broker (transform-offsets topic offsets-response conf)])))))))

(defn create-producers [broker-offsets conf]

	       (info "The record " topic " " partition " cannot be found"))))
    (p-close)))
     
(defn consume-producers! [conn 
                          bootstrap-brokers
                          group-conn
                          producers topics broker-offsets-p msg-ch {:keys [^Timer m-consume-cycle fetch-poll-ms] 
                                                                    :or {fetch-poll-ms 10000} :as conf}]
