   (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)
 

        start-offsets (:kafka/start-offsets task-map)]
    (cond checkpoint
          (do
           (info log-prefix "Seeking to checkpointed offset at:" checkpoint)
           (seek-to-offset! consumer {:topic topic :partition kpartition} checkpoint))

          start-offsets
          (let [offset (get start-offsets kpartition)]

  (:import (org.apache.kafka.clients.consumer ConsumerRecords ConsumerRecord)
           (org.apache.kafka.clients.consumer KafkaConsumer ConsumerRebalanceListener Consumer)
           (franzy.clients.consumer.client FranzConsumer)
           [java.util.concurrent.atomic AtomicLong]
           (org.apache.kafka.clients.producer Callback)
           [franzy.clients.producer.types ProducerRecord]
           [org.apache.kafka.common TopicPartition]))

(def defaults
  {:kafka/receive-buffer-bytes 65536
