Seek

A handler moves its running subscription through the SeekHandle key; the seek is local and a rebalance discards it.

Ctx<SeekHandle> binds a KafkaSeeker for the subscription the delivery came from. Its seek(position) takes a KafkaPosition, and the next delivery comes from there. Ctx<Partition> and Ctx<Position> give the delivery's own coordinates to aim with. A seek moves this consumer and no other member of the group. It commits nothing, and the next rebalance discards it: whoever gets the partitions resumes from the group's committed offsets. Under Commit::Tracked a seek clears the tracked position of every partition it moves, so a later commit never passes records the seek replayed and nobody handled. A seek after the service shut down returns an error. A batch handler reaches the same key through its KafkaBatchContext, and the move applies once the batch has settled. A position that has to survive a restart belongs in start_at instead.

Example

A job that carries resume_at moves its partition past a run of jobs known to fail.

Macros
Manual

examples/kafka_seek.rs
/// Skips a run of jobs known to fail: the partition jumps to `resume_at`.
#[subscriber(KafkaTopic::new("jobs").group("jobs-svc").commit(Commit::Tracked))]
async fn work(
    job: &Job,
    Ctx(partition): Ctx<Partition>,
    Ctx(seeker): Ctx<SeekHandle>,
) -> HandlerOutcome {
    tracing::info!(job.id, "worked");
    let Some(resume_at) = job.resume_at else {
        return HandlerOutcome::ack();
    };
    match seeker
        .seek(KafkaPosition::offset(partition, resume_at))
        .await
    {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

examples/manual/kafka_seek.rs
/// Skips a run of jobs known to fail: the partition jumps to `resume_at`.
struct Work;

impl Handle<Job, (), (), KafkaContext> for Work {
    async fn handle(
        &self,
        job: &Job,
        _outs: &(),
        ctx: &mut Context<'_, KafkaContext>,
    ) -> Result<(), HandlerOutcome> {
        tracing::info!(job.id, "worked");
        let Some(resume_at) = job.resume_at else {
            return Ok(());
        };
        let position = KafkaPosition::offset(ctx.context(Partition), resume_at);
        ctx.context(SeekHandle)
            .seek(position)
            .await
            .map_err(|_| HandlerOutcome::retry())
    }
}

fn app() -> impl App<State = ()> {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("jobs", "0.1.0")).with_broker(broker, |b| {
        let jobs = KafkaTopic::new("jobs")
            .group("jobs-svc")
            .commit(Commit::Tracked);
        b.include(subscriber("jobs", Work).build().map_source(|_| jobs));
    })
}

Example​

Example