               (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)]) (:offset v)))))
+		              (recur t (assoc state (clojure.string/join "/" [(:topic v) (:partition v)]) (inc (:offset v))))))
                ;timeout
               (do
                 (write-persister-data group-conn state)

  (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)]
    

									                                               broker topic partition 
									                                               (reentrant-lock group-conn (str topic "/" partition)))
																				                (rest partitions)))
									                               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
