 									                                               broker topic partition 
 									                                               (reentrant-lock group-conn (str topic "/" partition)))
 																				                (rest partitions)))
-									                               broker-offsets1))]
-        (info "add " add)
-        (info "broker-offsets1 " broker-offsets1)
-									    
-        (loop [broker-offsets2 broker-offsets1 partitions remove]
-          (if-let [record (first partitions)]
-            (let [{:keys [broker partition]} record]
-	            (recur
-	                 (do
-	                   (release group-conn (str topic "/" partition)) 
-	                   (change-partition-lock broker-offsets2 broker topic partition false)) 
-	                 (rest partitions)))
-	            broker-offsets2))))
+									                               broker-offsets1))
+         broker-offsets2
+							        (loop [broker-offsets2 broker-offsets1 partitions remove]
+							          (if-let [record (first partitions)]
+							            (let [{:keys [broker partition]} record]
+								            (recur
+								                 (do
+								                   (release group-conn (str topic "/" partition)) 
+								                   (change-partition-lock broker-offsets2 broker topic partition false)) 
+								                 (rest partitions)))
+								            broker-offsets2))]
+               
+           ;only allow offsets with locked true, this is done automatically by the change-partition-lock function
+          broker-offsets2
+        
+        ))
     
   
 (defn consume-producers! [bootstrap-brokers

  (persistent-set* group-conn (vec state)))
  
(defn get-persister [group-conn conf]
  (let [{:keys [offset-commit-freq] :or {offset-commit-freq 5000}} conf
        ch (chan 100)]
    

              (if (nil? v)
                 (write-persister-data group-conn state) ;channel is closed
		            (if (= c ch)
		              (recur t (assoc state (clojure.string/join "/" [(:topic v) (:partition v)]) (inc (:offset v))))))
               ;timeout
              (do
                (write-persister-data group-conn state)
