    (merge-broker-offsets broker-offsets v))
 
 
-(defn close-and-reconnect [bootstrap-brokers producers topics errors conf]
+(defn close-and-reconnect [conn bootstrap-brokers producers topics errors conf]
   
   (let [reconnect (some (fn [x] (not (instance? FetchError x))) errors)]
     

	       (info "The record " topic " " partition " cannot be found"))))
    (p-close)))
     
(defn consume-producers! [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}]

			        (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 (get-create-offset-producer offset-producers broker conf)
			                offsets-response (get-offsets offset-producer topic (map first v))]
			            [broker (transform-offsets topic offsets-response conf)])))))))

(defn create-producers [broker-offsets conf]
