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

							         (let [k #{(:topic msg) (:partition msg)}]
							           (if (is-new-msg? current-offsets resp k msg)   
				                   (do 
                               (>!! msg-ch msg)
		                           (p-send msg)
                               (.mark m-consume-reads) ;metrics mark

							         (do (error "Fetch error: " msg) (tuple resp (conj errors msg)))
							         :else (throw (RuntimeException. (str "The message type " msg " not supported")))))
		               (catch Exception e 
	                  (do (.printStackTrace e)
                        (prn-fetch-error e state msg)
	                      (tuple resp errors))
	                  )))
                  (do (error "State not supported " state)
