                         :message (deserializer-fn (.value cr))
                         :offset (.offset cr)})
                      (fn [^ConsumerRecord cr]
-                       (deserializer-fn (.value cr))))]
+                       (deserializer-fn (.value cr))))
+        read-offset (:read-offset monitoring)]
     (->KafkaReadMessages log-prefix task-map topic nil batch-timeout
-                         deserializer-fn segment-fn nil nil nil false)))
+                         deserializer-fn segment-fn read-offset nil nil nil false)))
 
 (defn close-read-messages
   [event lifecycle]

                                  batch-timeout)))
          nil))))

(defn read-messages [{:keys [onyx.core/task-map onyx.core/log-prefix] :as event}]
  (let [{:keys [kafka/topic kafka/deserializer-fn]} task-map
        batch-timeout (arg-or-default :onyx/batch-timeout task-map)
        wrap-message? (or (:kafka/wrap-with-metadata? task-map) (:kafka/wrap-with-metadata? defaults))

    (if (and iter (.hasNext ^java.util.Iterator iter))
      (let [rec ^ConsumerRecord (.next ^java.util.Iterator iter)
            new-offset (if rec (.offset rec) offset)
            deserialized (some-> rec segment-fn)]
        (if (= :done deserialized)
          (do (set! drained true)
              nil)
          (do
           (.set ^AtomicLong read-offset new-offset)
           (set! offset new-offset) 
           deserialized))) 
      (do (set! iter 
