   (if (:kafka/force-reset? task-map)
     (read-from-bound consumer topic kpartition task-map)
     (if-let [x (kzk/committed-offset m group-id topic kpartition)]
-      x
+      (inc x)
       (read-from-bound consumer topic kpartition task-map))))
 
 (defn highest-offset-to-commit [offsets]

    [_ _ message-id]
    (when-let [msg (get @pending-messages message-id)]
      (swap! pending-messages dissoc message-id)
      (>!! read-ch msg)))

  (pending?
    [_ _ message-id]

                  (Thread/sleep empty-read-back-off)
                  (recur fetched head-offset))
                (let [next-offset (.offset ^int (first ms))]
                  (>!! ch (assoc (t/input (java.util.UUID/randomUUID)
                                          (deserializer-fn (.value ^KafkaMessage (first ms))))
                                 :offset next-offset))
                  (recur (rest ms) (inc next-offset)))))
            (finally
             (future-cancel commit-fut))))
