  (defn get-partitions-to-lock [topic broker-offsets members]
    "broker-offsets {broker {topic [{:partition :offset :topic}]}}
     Returns the number of partitions that should be locked"
+   ;;TODO the problem is that the partitions that were marked as locked are not locked in broker-offsets, i.e. the locked
+   ;;attribute is not kept in the map
    (try
 	   (let [broker-partitions (filter #(= (:topic %) topic) (flatten-broker-partitions broker-offsets))
 	         partition-count (count broker-partitions)

			         (error "Error close and reconnect: " errors)
            
			         (let [[producers broker-offsets] (close-and-reconnect conn bootstrap-brokers producers topics errors conf)]
	                ;;here we need to delete the offsets that have had errors from the storage
	                ;;or better yet set them to storage
                 ;persist-error-offsets [group-conn broker-offsets errors conf]
                  (info "Got new consumers " (map :broker producers))
	                (persist-error-offsets group-conn broker-offsets errors conf)
                  (.stop timer-ctx)
			            (recur producers broker-offsets)))

   "
  ;(info "merge " state " with " d)
  ;(prn "state d " d) 
  ;(clojure.pprint/pprint curr-state)
  (let [r (reduce (fn [state [broker messages]]
                    (reduce (fn [state {:keys [topic partition offset error-code]}]
                              ;note that the record fomr which the keys are drawn does not have the locked key
                              ;(info "$$$$ current record " (get-current-partition broker topic partition state))
                              (merge-with merge
                                 state
			                           (if (or (not error-code) (= error-code 0))
                                   {broker
			                                 {topic
			                                      (conj (get-rest-of-partitions broker topic partition state)
			                                            {:offset (inc offset) :locked (:locked (get-current-partition broker topic partition state))
                                                   :partition partition :error-code (if error-code error-code 0)  })
                                         }
                                      })))
                            curr-state messages))
          curr-state d)]
    ;(clojure.pprint/pprint ["r " r])
    r))

(defn- get-latest-offset [k current-offsets resp]
