   "Returns [the messages, and fetch errors], if any error was or timeout was detected the function returns otherwise it waits for a FetchEnd message
    and returns. 
   "
+  ;(info "!!!! calling get-locked-partitions with " topic-offsets)
   (let [locked-partitions (get-locked-partitions topic-offsets)]
       (do
-	      ;(info "!!!!!!send fetch " (:broker producer) " "  locked-partitions)
+	      (info "!!!!!!send fetch " (:broker producer) " "  locked-partitions)
 			  (send-fetch producer locked-partitions)
 			  
 			  (let [

	         e (long (/ partition-count member-count))
	         l (rem partition-count member-count)]
	     
	     (info "members " members " partition-count " partition-count " locked-partition-count " locked-partition-count " e " e " l " l )
	     [(if (> e locked-partition-count) (count (get-add-partitions broker-partitions e)) 0)
	      (if (> locked-partition-count e) (count (get-remove-partitions broker-partitions (- locked-partition-count e))) 0)
	      l

  
  (loop [producers producers broker-offsets1 broker-offsets-p]
      ;v is [broker data]
       ;(info "11 broker-offsets1 " broker-offsets1)
	      (let [ broker-offsets2 (apply merge-with merge 
	                                            (for [topic topics] (calculate-locked-offsets topic group-conn 
                                                                   ;;we need to remove all other topics from the offset map
                                                                   (into {} (for [[broker topics] broker-offsets1  
                                                                                  [topic1 offsets] topics :when (= topic1 topic) ] [broker {topic offsets}]))
                                                                   conf)))
               timer-ctx (.time m-consume-cycle)
               q (consume-brokers! producers group-conn broker-offsets2 msg-ch conf)]
         
         
	       (let [[v errors] q]
			    (if (and (not (nil? errors)) (> (count errors) 0))
