    and added to the complete-queue queue.
    Returns the state map with the :status and :producers updated
   "
-  [{:keys [redis-conn producers work-queue working-queue complete-queue conf prev-offsets-read] :as state} work-unit f-delegate]
+  [{:keys [redis-conn producers work-queue working-queue complete-queue conf] :as state} work-unit f-delegate]
   ;(prn "wait-and-do-work-unit! >>> have work unit " work-unit)
   ;(info "prev-offsets-read: " prev-offsets-read)
   (io!

(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-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}}]
  ;(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))
      error-ch (handle-error-response v)
      (handle-timeout-response prev-offset))))

    
(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}}]
    (io!
      (send-fetch producer [[topic [{:partition partition :offset offset}]]])
      (handle-response producer work-unit f-delegate (get state :conf) :prev-offset prev-offset)))


(defn- safe-sleep


              (assoc
                  state2
                ;:prev-offsets-read (assoc-in (if prev-offsets-read prev-offsets-read {}) [topic partition] prev-offset)
                :producers producers2)))
          (catch Throwable t (do
                               (.printStackTrace t)
