 								         (do
                            (if (and (= (:topic msg) topic) (= ^Long (:partition msg) ^Long partition)
                                     (>= ^Long (:offset msg) offset)
-                                    (< ^Long (:offset msg) max-offset) (< ^Long prev-offset1 (:offset msg)))
+                                    (< ^Long (:offset msg) max-offset)
+                                    ;(< ^Long prev-offset1 (:offset msg))
+                                    )
                              (tuple (conj resp msg) errors (f-delegate f-state msg) (:offset msg))
-                             (tuple resp errors f-state prev-offset1)))
+                             (tuple resp errors f-state)))
 								         (instance? FetchError msg)
-								         (do (error "Fetch error: " msg) (tuple resp (conj errors msg) f-state prev-offset1))
+								         (do (error "Fetch error: " msg) (tuple resp (conj errors msg) f-state))
 								         :else (throw (RuntimeException. (str "The message type " msg " not supported")))))
 			               (catch Exception e
 		                  (do (error e e)
-		                      (tuple resp errors f-state prev-offset1))
+		                      (tuple resp errors f-state))
 		                  )))
 	                  (do (error "State not supported " state)
-	                      [{} [] nil prev-offset])
+	                      [{} [] nil])
 	                  )))]
          ;(info "FETCH RESP " fetch-res)
 	       (if (coll? fetch-res)


(defn read-fetch-message 
  "read-fetch will return the result of fn which is [resp-vec error-vec]"
  [{:keys [topic partition ^Long offset ^Long len]} f-delegate v & {:keys [prev-offset] :or {prev-offset -1}}]
  (if (byte-array? v)
	  (let [ ^Long max-offset (+ offset len)
	         fetch-res
	         (read-fetch (Unpooled/wrappedBuffer ^"[B" v) [[] [] nil prev-offset]
				     (fn [state msg]
	              ;read-fetch will navigate the fetch response calling this function
               ;(info "READ FETCH MESSAGE " msg)
	              (if (coll? state)
			            (let [[resp errors f-state prev-offset1] state]
		               (try
			               (do
					             (cond

        (try
          (do
            (if (not producer-conn) (throw (RuntimeException. "No producer created")))
            (let [[status resp-data] (fetch-and-wait state work-unit producer-conn f-delegate)
                  state2 (assoc state :status :ok)
                  ]
              (if resp-data
