                                (not (:kafka/partition task-map)))
         _ (check-num-peers-equals-partitions task-map n-partitions)
         _ (assign-partitions! consumer [{:topic topic :partition kpartition}])
-        _ (seek-offset! log consumer group-id topic kpartition task-map)
+        _ (seek-offset! (:log-prefix compiled) log consumer group-id topic kpartition task-map)
         offset (cp/next-offset consumer {:topic topic :partition kpartition})
         commit-interval (or (:kafka/commit-interval task-map) (:kafka/commit-interval defaults))
         commit-fut (future (commit-loop log group-id topic kpartition commit-interval pending-commits))]

        (throw e)))))

(defn start-kafka-consumer
  [{:keys [onyx.core/task-map onyx.core/pipeline onyx.core/log onyx.core/replica] :as event} lifecycle]
  (let [{:keys [kafka/topic kafka/partition kafka/group-id kafka/consumer-opts]} task-map
        brokers (find-brokers (:kafka/zookeeper task-map))
        commit-interval (or (:kafka/commit-interval task-map) (:kafka/commit-interval defaults))

(defn checkpoint-name [group-id topic assigned-partition]
  (format "%s-%s-%s" group-id topic assigned-partition))

(defn checkpoint-or-beginning [log-prefix log consumer topic kpartition group-id task-map]
  (let [k (checkpoint-name group-id topic kpartition)]
    (try
      (let [offset (inc (:offset (read-commit log k)))]
        (info log-prefix "Seeking to checkpointed offset at:" offset)
        (seek-to-offset! consumer {:topic topic :partition kpartition} offset))
      (catch org.apache.zookeeper.KeeperException$NoNodeException nne
        (try
