     (reentrant-lock group-conn host (str topic "/" partition))
     (reentrant-lock group-conn (str topic "/" partition))))
 
-(defn calculate-locked-offsets [topic group-conn broker-offsets conf]
+
+
+(defn calculate-locked-offsets [topic group-conn init-broker-offsets conf]
   "broker-offsets have format  {broker {topic [{:partition :offset :topic}]}}
    calculate which offsets should be consumed based on the locks and other members
    returns the broker-offsets marked as locked or not as locked."
   ;(info "host " (get conf :host-name))
+  
   (let [
-        broker-partitions (filter #(= (:topic %) topic) (flatten-broker-partitions broker-offsets))
-        [locked-n remove-n l] (get-partitions-to-lock topic broker-offsets (get-members group-conn))
+        broker-partitions (filter #(= (:topic %) topic) (flatten-broker-partitions init-broker-offsets))
+        [locked-n remove-n l] (get-partitions-to-lock topic init-broker-offsets (get-members group-conn))
         
-          broker-offsets1 (loop [broker-offsets1 broker-offsets locked-i locked-n  remove-i remove-n l-i l partitions broker-partitions]
+          broker-offsets2 (loop [broker-offsets1 init-broker-offsets locked-i locked-n  remove-i remove-n l-i l partitions broker-partitions]
                             (if-let [record (first partitions)]
-                              (let [{:keys [broker partition locked]} record]
+                              (let [{:keys [topic broker partition locked]} record]
                                  (cond 
                                    (and locked (> remove-i 0))
                                    (do 

  "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. 
  "
  (let [locked-partitions (get-locked-partitions topic-offsets)]
      (do
	      ;(info "!!!!!!send fetch " (:broker producer) " "  locked-partitions)
			  (send-fetch producer locked-partitions)
			  
			  (let [

		                                                                                                    (:offset p-record) ) ))
                                                   }})
          (do
            (error "Error no record found : " p-record)
            broker-offsets)
          
        )))
