 (defn handle-error-response [v]
   [:fail v])
 
-(defn handle-read-response [work-unit f-delegate v & {:keys [prev-offset] :or {prev-offset -1}}]
-  (let [[resp-vec error-vec _ prev-offset1] (read-fetch-message work-unit f-delegate v :prev-offset prev-offset)]
-    [:ok resp-vec prev-offset1]))
-
-(defn handle-timeout-response [prev-offset]
-  [:fail nil prev-offset])
+(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 & {:keys [prev-offset] :or {prev-offset -1}}]
+  [{: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 :prev-offset prev-offset)
-                (instance? Poison v) [:fail nil prev-offset]
-                :else (handle-read-response work-unit f-delegate v :prev-offset prev-offset))
+                (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 prev-offset))))
+      (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 & {:keys [prev-offset] :or {prev-offset -1}}]
+  [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) :prev-offset prev-offset)))
+      (handle-response producer work-unit f-delegate (get state :conf))))
 
 
 (defn- safe-sleep

								         (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)))
                             (tuple (conj resp msg) errors (f-delegate f-state msg) (:offset msg))
                             (tuple resp errors f-state prev-offset1)))
								         (instance? FetchError msg)
								         (do (error "Fetch error: " msg) (tuple resp (conj errors msg) f-state prev-offset1))
								         :else (throw (RuntimeException. (str "The message type " msg " not supported")))))
			               (catch Exception e
		                  (do (error e e)
		                      (tuple resp errors f-state prev-offset1))
		                  )))
	                  (do (error "State not supported " state)
	                      [{} [] nil prev-offset])
	                  )))]
         ;(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]
  (if (byte-array? v)
	  (let [ ^Long max-offset (+ offset len)
	         fetch-res
	         (read-fetch (Unpooled/wrappedBuffer ^"[B" v) [[] [] nil -1]
				     (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] state]
		               (try
			               (do
					             (cond
