          (info "No messages consumed " fetch-res)
          nil))))
   
+
+(defn- get-locked-partitions [topic-offsets]
+  "Get the locked partitions"
+  (map (fn [[k v]]
+         (tuple k
+           (filter
+             :locked
+             v)))
+    topic-offsets))
+             
 (defn send-request-and-wait [producer group-conn topic-offsets msg-ch {:keys [^Histogram m-message-size
                                                                               ^Meter m-consume-reads fetch-timeout] 
                                                                        :or {fetch-timeout 60000} :as conf}]
   "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 "!!!!!!send fetch " (:broker producer) " "  (map (fn [[k v]] [k v]) topic-offsets))
-  (send-fetch producer (map (fn [[k v]] [k v]) topic-offsets))
-  
-  (let [
-        persister (get-persister group-conn conf)
-        {:keys [read-ch error-ch]} (:client producer)
-        current-offsets (into {} (for [[topic v] topic-offsets
-                                        msg   v]
-                                      [#{topic (:partition msg)} (assoc msg :topic topic) ]))]
-    
-      (let [[v c] (alts!! [read-ch error-ch (timeout fetch-timeout)])]
-        ;(info "Got message " (count v ) " is read " (= c read-ch) " is error " (= c error-ch))
-        (try
-	        (cond 
-	          (= c read-ch)
-            ;;read-fetch will navigate and process the fetch response, sending messages to msg-ch
-	          (read-fetch-message persister current-offsets msg-ch m-consume-reads m-message-size v)
-	          (= c error-ch)
-	          (do 
-	            (error v v)
-	            [[] [{:error v}]])
-	          :else
-	          (do 
-	            (error "timeout reading from " (:broker producer))
-	            [[] [{:error (RuntimeException. (str "Timeout while waiting for " (:broker producer)))}]] 
-	            ))
-         (finally ((:p-close persister)))))))
+  (let [locked-partitions (get-locked-partitions topic-offsets)]
+      (do
+	      (info "!!!!!!send fetch " (:broker producer) " "  locked-partitions)
+			  (send-fetch producer locked-partitions)
+			  
+			  (let [
+			        persister (get-persister group-conn conf)
+			        {:keys [read-ch error-ch]} (:client producer)
+			        current-offsets (into {} (for [[topic v] topic-offsets
+			                                        msg   v]
+			                                      [#{topic (:partition msg)} (assoc msg :topic topic) ]))]
+			    
+			      (let [[v c] (alts!! [read-ch error-ch (timeout fetch-timeout)])]
+			        ;(info "Got message " (count v ) " is read " (= c read-ch) " is error " (= c error-ch))
+			        (try
+				        (cond 
+				          (= c read-ch)
+			            ;;read-fetch will navigate and process the fetch response, sending messages to msg-ch
+				          (read-fetch-message persister current-offsets msg-ch m-consume-reads m-message-size v)
+				          (= c error-ch)
+				          (do 
+				            (error v v)
+				            [[] [{:error v}]])
+				          :else
+				          (do 
+				            (error "timeout reading from " (:broker producer))
+				            [[] [{:error (RuntimeException. (str "Timeout while waiting for " (:broker producer)))}]] 
+				            ))
+			         (finally ((:p-close persister)))))))))
 
 
 (defn consume-broker [producer group-conn topic-offsets msg-ch conf]

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

          
        )))

(defn- release-partition [group-conn topic partition conf]
  (if-let [host (get conf :host-name nil)] 
    (release group-conn host (str topic "/" partition))
    (release group-conn (str topic "/" partition))))
    
(defn- lock-partition [group-conn topic partition conf]
  (if-let [host (get conf :host-name nil)]  
    (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]
  "broker-offsets have format  {broker {topic [{:partition :offset :topic}]}}
   calculate which offsets should be consumed based on the locks and other members
