 								         (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

(defn handle-error-response [v]
  [:fail v])

(defn handle-read-response [work-unit f-delegate v]
  (let [[resp-vec error-vec] (read-fetch-message work-unit f-delegate v)]
    [:ok resp-vec]))

(defn handle-timeout-response []
  [:fail nil])


(defn handle-response
  "Listens to a response after a fetch request has been sent
   Returns [status data]  status can be :ok, :timeout, :error and data is v returned from the channel"
  [{:keys [client] :as state} work-unit f-delegate conf]
  ;(prn "handler-response >>>>> " work-unit)
  (let [fetch-timeout (get conf :fetch-timeout 10000)
        {:keys [read-ch error-ch]} client
        [v c] (alts!! [read-ch error-ch (timeout fetch-timeout)])]
    (condp = c
      read-ch (cond 
                (instance? Reconnected v) (handle-response state f-delegate conf)
                (instance? Poison v) [:fail nil]
                :else (handle-read-response work-unit f-delegate v))
      error-ch (handle-error-response v)
      (handle-timeout-response))))

    
(defn fetch-and-wait 
  "
   Sends a request for the topic and partition to the producer
   Returns [status data]  status can be :ok, :fail and data is v returned from the channel"
  [state {:keys [topic partition offset len] :as work-unit} producer f-delegate]
    (io!
      (send-fetch producer [[topic [{:partition partition :offset offset}]]])
      (handle-response producer work-unit f-delegate (get state :conf))))


(defn- safe-sleep
