 (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,
+   The function will merge d with state so that state will contain the latest offsets d,
    and then returns the new state
    "
   ;(info "merge " state " with " d)

                   (doseq [c consumers]
                     ((:shutdown c))))]
    
    {:shutdown shutdown :message-ch msg-ch :group-conn group-conn :metrics metrics}))


(defn close-consumer [{:keys [shutdown]}]

	                ;;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)))
			      (do
               (if (< (reduce #(+ %1 (count %2)  ) 0 (vals v)) 1) ; if we were reading data, no need to pause
	               (do (info "sleep: " fetch-poll-ms) (<!! (timeout fetch-poll-ms))))
	             
               (.stop timer-ctx)
