   "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 [

                                           
                                 broker-offsets1))]
        
					
	          ;(clojure.pprint/pprint broker-offsets1)  
	          broker-offsets1
	          
	        ))
    

	         (p-send {:topic topic :partition partition :offset (:offset record)}))
	       (info "The record " topic " " partition " cannot be found"))))
    (p-close)))

     
(defn consume-producers! [conn 
                          bootstrap-brokers
