KafkaBatchContext

A batch handler's context, KafkaBatchContext, carries the subscription's seek handle; each record's own coordinates travel with the elements.

A batch spans many records, so it has no single topic, partition or offset. The subscription builds the context once per batch, and it holds what every record of the batch shares: the SeekHandle key, the same key a KafkaContext carries. A batch body takes ctx: &mut Context<'_, KafkaBatchContext> and reads the handle with ctx.context(SeekHandle). A seek from a batch body takes effect once the whole batch has settled, so the next batch opens at the new position. A per-record key such as Partition or Offset is a compile error in a batch body, because the batch context does not carry it. A batch registration has no per-record topic for ToSourceTopic either, so it names a fixed retry topic.

Example

Macros
Manual

examples/kafka_context_batch.rs
/// The batch settles first; the seek then opens the next batch at the offset a job named, so
/// the run from there is processed again.
#[subscriber(KafkaTopic::new("jobs").group("jobs-svc").commit(Commit::Tracked))]
async fn drain(
    jobs: &[Job],
    ctx: &mut Context<'_, KafkaBatchContext>,
    State(redos): State<Arc<Redos>>,
) -> HandlerOutcome {
    tracing::info!(jobs = jobs.len(), "draining jobs");
    let target = jobs
        .iter()
        .filter(|job| redos.first_time(job))
        .find_map(|job| job.redo_from);
    if let Some((partition, offset)) = target
        && ctx
            .context(SeekHandle)
            .seek(KafkaPosition::offset(partition, offset))
            .await
            .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/kafka_context_batch.rs
/// The batch settles first; the seek then opens the next batch at the offset a job named, so
/// the run from there is processed again.
struct Drain;

impl Handle<[Job], (), (), KafkaBatchContext, AppState> for Drain {
    async fn handle(
        &self,
        jobs: &[Job],
        _outs: &(),
        ctx: &mut Context<'_, KafkaBatchContext, AppState>,
    ) -> Result<(), Vec<HandlerOutcome>> {
        tracing::info!(jobs = jobs.len(), "draining jobs");
        let redos = &ctx.state().redos;
        let target = jobs
            .iter()
            .filter(|job| redos.first_time(job))
            .find_map(|job| job.redo_from);
        if let Some((partition, offset)) = target
            && ctx
                .context(SeekHandle)
                .seek(KafkaPosition::offset(partition, offset))
                .await
                .is_err()
        {
            return Err(jobs.iter().map(|_| HandlerOutcome::retry()).collect());
        }
        Ok(())
    }
}

fn jobs() -> KafkaTopic {
    KafkaTopic::new("jobs")
        .group("jobs-svc")
        .commit(Commit::Tracked)
}

Example​

Example