     (reentrant-lock group-conn (str topic "/" partition))))
 
 
-;global cached offsets for a topic
-(def cached-offsets (ref {}))
 
 (defn calculate-locked-offsets [topic group-conn init-broker-offsets conf cached-offsets]
   (let [

            [group-redis.core :refer [host-name]]
            [group-redis.partition :refer [controlled-assignments]]
            [clojure.pprint :refer [pprint]]
            [clojure.core.async :refer [<!! >!! alts!! timeout chan go >! <! close! go-loop]]
            [clj-tuple :refer [tuple]])
  (:import [kafka_clj.fetch Message FetchError]

    (if-let [producer (first (filter (fn [{:keys [broker]}] (= broker broker-k)) producers))]
      producer
      (create-fetch-producer broker-k conf))))

(defn- parallel-broker-consume [group-conn broker-offsets msg-ch conf {:keys [broker] :as producer}]
  (info "inside parallel broker consume")
  (vector broker  (consume-broker producer group-conn (get broker-offsets broker) msg-ch conf)))

(defn ^Callable callable 
  "Returns a function as a Callable that applies (f i)"
  [f i] 
  (fn [] (f i) ))

(defn pmap-exec 
  "Calls (f i) inside a thread where i is a item in the list l for each i in l, this returns a list of Future
   Then calls .get on each future, the result is a lazy list of results from futures that needs to be run in reduce or doall"
  [^ExecutorService exec f l]
  (r/map (fn [^Future v] (.get v 5 java.util.concurrent.TimeUnit/MINUTES))
    (doall (map (fn [i] (.submit exec (callable f i))) l))))

(defn consume-brokers! [producers exec group-conn broker-offsets msg-ch conf]
  "
   Broker-offsets should be {broker {topic [{:offset o :partition p} ...] }}
   Consume brokers and returns a list of lists that contains the last messages consumed, or -1 -2 where errors are concerned
   the data structure returned is {broker -1|-2|[{:offset o topic: a} {:offset o topic a} ... ] ...}
  "
  (try
    (r/fold 
      (fn
        ([] [{} []])
        ([[state errors] [broker [msgs msg-errors]]]
         [(merge state {broker msgs}) (if (> (count msg-errors) 0) (apply conj errors msg-errors) errors)]))
          (pmap-exec exec
            (partial parallel-broker-consume group-conn  broker-offsets msg-ch conf)
                    producers))
   (finally
     ;(info ">>>>>>>>>>>>>>>>>>>>> END CONSUME BROKERS!")
     )))


(defn update-broker-offsets [broker-offsets v]
  "
