   (let [policy (:kafka/offset-reset task-map)
         start-offsets (:kafka/start-offsets task-map)]
     (doseq [kpartition kpartitions]
-      (cond checkpoint
-            (do
-              (info log-prefix "Seeking to checkpointed offset at:" (inc checkpoint))
-              (seek-to-offset! consumer {:topic topic :partition kpartition} (inc checkpoint)))
+      (cond (get checkpoint kpartition)
+            (let [offset (get checkpoint kpartition)]
+              (info log-prefix "Seeking to checkpointed offset at:" (inc offset))
+              (seek-to-offset! consumer {:topic topic :partition kpartition} (inc offset)))
 
             start-offsets
             (let [offset (get start-offsets kpartition)]

(deftype KafkaReadMessages 
  [log-prefix task-map topic ^:unsynchronized-mutable kpartitions batch-timeout
   deserializer-fn segment-fn read-offset ^:unsynchronized-mutable consumer 
   ^:unsynchronized-mutable iter ^:unsynchronized-mutable offset ^:unsynchronized-mutable drained]
  p/Plugin
  (start [this event]
    (let [{:keys [kafka/group-id kafka/consumer-opts]} task-map


  p/Checkpointed
  (checkpoint [this]
    partition->offset)

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

