 	                                            (calculate-locked-offsets topic group-conn broker-offsets1 conf)))
                timer-ctx (.time m-consume-cycle)
                q (consume-brokers! producers group-conn broker-offsets2 msg-ch conf)]
-	       (let [[v errors] q
-               errors2 (filter #(> (:error-code %) 1) errors) ;exclude out of range errors
-               ]
-			    (if (and (not (nil? errors2)) (> (count errors2) 0))
+	       (let [[v errors] q]
+			    (if (and (not (nil? errors)) (> (count errors) 0))
 			      (do
 			         (error "Error close and reconnect: " errors)
             
-			         (let [[producers broker-offsets] (close-and-reconnect bootstrap-brokers producers topics errors conf)]
+			         (let [[producers broker-offsets] (close-and-reconnect conn 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]

	
	  (info "close-and-reconnect: " bootstrap-brokers " topic " topics " reconnect " reconnect)
	  (if-let [metadata (get-metadata bootstrap-brokers conf)]
	    (let [broker-offsets (doall (get-broker-offsets metadata topics conf))
	          producers (if reconnect (doall (create-producers broker-offsets conf)) producers)]
	      [producers broker-offsets])
	    (throw (RuntimeException. "No metadata from brokers " bootstrap-brokers)))))

	       (info "The record " topic " " partition " cannot be found"))))
    (p-close)))
     
(defn consume-producers! [conn 
                          bootstrap-brokers
                          group-conn
                          producers topics broker-offsets-p msg-ch {:keys [^Timer m-consume-cycle fetch-poll-ms] 
                                                                    :or {fetch-poll-ms 10000} :as conf}]
