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

	       (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

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