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

  (let [
        metrics (create-metrics)
        msg-ch (chan 100)
        msg-buff (buffered-chan msg-ch 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 [[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]
