                    (doseq [c consumers]
                      ((:shutdown c))))]
     
-    {:shutdown shutdown :message-ch msg-ch :group-conn group-conn :metrics metrics}))
+    {:shutdown shutdown :message-ch msg-ch :group-conn group-conn :metrics metrics :consumers consumers}))
 
 
 (defn close-consumer [{:keys [shutdown]}]

  "Consume from the current offsets,
   if any error the producers are closed and a reconnect is done, and consumption is tried again
   otherwise the broker-offsets are updated and the next fetch is done"
  (loop [producers producers broker-offsets1 broker-offsets-p]
      ;v is [broker data]
	      (let [ broker-offsets2 (apply merge-with merge 

                                        msg   v]
                                      [#{topic (:partition msg)} (assoc msg :topic topic) ]))]
    
    (loop [resp {} fetch-errors [] t (timeout fetch-timeout)]
      (let [[v c] (alts!! [read-ch error-ch t])]
        (.mark m-consume-reads) ;metrics mark
        (if v
		      (if (= c read-ch)
		        (cond (instance? FetchEnd v)  (do (p-close) [(vals resp) fetch-errors])
                  :else ;assume FetchMessage
                  (do
                     (let [k #{(:topic v) (:partition v)}]
                       
                       (if (instance? FetchError v)
                         (do
                           (error "Fetch error " v)
                           (recur resp (conj fetch-errors v) (timeout fetch-timeout)))
                         
                         (if-let [partition (:partition v)] 
                           (let [latest-offset (get-latest-offset k current-offsets resp)
                                 new-msg? (or (> (:offset v) latest-offset) (= (:offset v) 0)) ]
		                         (if new-msg?
			                          (do 
                                   (.update m-message-size (count (:bts v)))
		                               (>!! msg-ch v)
		                               (p-send v)))
                                                      
		                         (recur (if new-msg? (assoc resp k v) resp) fetch-errors (timeout fetch-timeout)))
                           (error "No partition sent " v)
                           )))))
		        (do (p-close) (error "Error while requesting data from " (:broker producer) " for topics " (keys topic-offsets)) [(vals resp) (conj fetch-errors v)]))
		        (do 
                (p-close) 
                (error "Timeout while requesting data from " (:broker producer) " read messages: " resp) 
                [(vals resp) (conj fetch-errors (RuntimeException. (str "Timeout while waiting for send response from " (:broker producer) " topic ")))]))))
    
    
    ))
