   "Consume from the current offsets,
    if any error the producers are closed and a reconnect is done, and consumption is tried again
    otherwise the broker-offsets are updated and the next fetch is done"
+  
   (loop [producers producers broker-offsets1 broker-offsets-p]
       ;v is [broker data]
 	      (let [ broker-offsets2 (apply merge-with merge 

(defn merge-broker-offsets [state d]
  "D is a collection of messages one per topic partition, that were last consumed from a fetch request,
   state is the broker-offsets {broker {topic [{:partition :offset :topic}]}}
   The function will merge d with state so that state will contain the latest offsets consumed,
   and then returns the new state
   "
  ;(info "merge " state " with " d)

            [kafka-clj.metadata :refer [get-metadata]]
            [fun-utils.core :refer [buffered-chan]]
            [clojure.pprint :refer [pprint]]
            [clojure.core.async :refer [<!! >!! alts!! timeout chan go >! <! close! go-loop]])
  (:import [kafka_clj.fetch_codec FetchMessage FetchError FetchEnd]
           [com.codahale.metrics Meter MetricRegistry Timer Histogram]
           [java.util.concurrent Executors ExecutorService Future Callable]))
