                         :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]


(deftype KafkaReadMessages 
  [log-prefix task-map topic ^:unsynchronized-mutable kpartition batch-timeout
   deserializer-fn segment-fn ^:unsynchronized-mutable consumer 
   ^:unsynchronized-mutable iter ^:unsynchronized-mutable offset ^:unsynchronized-mutable drained]
  p/Plugin
  (start [this event]

  (recover! [this replica-version checkpoint]
    (set! drained false)
    (set! iter nil)
    (set! offset checkpoint)
    (seek-offset! log-prefix consumer kpartition task-map topic checkpoint)
    this)

