   (let [
         metrics (create-metrics)
         msg-ch (chan 100)
-        msg-buff (buffered-chan msg-ch 1000 1000)
+        msg-buff (buffered-chan msg-ch 1000 1000 1000)
         redis-conf (get conf :redis-conf {:heart-beat-freq 10})
         group-conn (let [c (create-group-connector (get redis-conf :redis-host "localhost") redis-conf)
                          host-name (get conf :host-name nil) ]

	       (do
	         (info "updating " topic " " partition " to " record)
	         (p-send {:topic topic :partition partition :offset (:offset record)}))
	       (error "The record " topic " " partition " cannot be found"))))
    (p-close)))
     
(defn consume-producers! [bootstrap-brokers

	       (let [[v errors] q]
			    (if (> (count errors) 0)
			      (do
			         (error "Error close and reconnect: " errors)
            
			         (let [[producers broker-offsets] (close-and-reconnect 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]
