                                         msg   v]
                                       [#{topic (:partition msg)} (assoc msg :topic topic) ]))]
     
-    (loop [resp {} fetch-errors {} t (timeout fetch-timeout)]
+    (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 
-                                         
-	                                         ;(if (= (dec fetch-count-i) 0)
-	                                             (do (p-close) [(vals resp) fetch-errors]))
-	                                          ;   (recur resp fetch-errors (timeout fetch-timeout) (dec fetch-count-i))))
+		        (cond (instance? FetchEnd v)  (do (p-close) [(vals resp) fetch-errors])
                   :else ;assume FetchMessage
-                     (do
+                  (do
+                     (let [k #{(:topic v) (:partition v)}]
+                       
                        (if (instance? FetchError v)
                          (do
                            (error "Fetch error " v)
-                           (recur resp (assoc k v fetch-errors) (timeout fetch-timeout)))
+                           (recur resp (conj fetch-errors v) (timeout fetch-timeout)))
                          
                          (if-let [partition (:partition v)] 
-                           (let [k #{(:topic v) partition}
-                             latest-offset (get-latest-offset k current-offsets resp)
-                             new-msg? (or (> (:offset v) latest-offset) (= (:offset v) 0)) ]
+                           (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)))
-		                            ;;(error "Duplicate message " k " latest-offset " latest-offset " message offset " (:offset v)))
-                           
-		                       (recur (if new-msg? (assoc resp k v) resp) fetch-errors (timeout fetch-timeout)))
+                                                      
+		                         (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 " producer " for topics " (keys topic-offsets)) [(vals resp) fetch-errors]))
+                           )))))
+		        (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) fetch-errors]))))
+                (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 ")))]))))
     
     
     ))

                   (doseq [c consumers]
                     ((:shutdown c))))]
    
    {:shutdown shutdown :message-ch msg-ch :group-conn group-conn :metrics metrics}))


(defn close-consumer [{:keys [shutdown]}]

(defn merge-broker-offsets [state d]
  "D is a collection of messages one per topic partition, that were last consumed from a fetch request,
   state is the broker-offsets {broker {topic [{:partition :offset :topic}]}}
   The function will merge d with state so that state will contain the latest offsets d,
   and then returns the new state
   "
  ;(info "merge " state " with " d)
