KafkaContext

KafkaContext gives a handler the record's topic, partition, offset, timestamp and key, each read as a parameter through Ctx<K>.

The subscription builds it once per delivery, beside the payload and the headers. A handler binds one field with Ctx<K>, where K is a key of context::keys:

Key	Value	What it is
`Topic`	`Str`	the topic the record was read from, which differs per record on a topic list or pattern
`Partition`	`i32`	the partition the record was read from
`Offset`	`i64`	the record's offset in its partition
`TimestampMillis`	`Option<i64>`	the record's timestamp in milliseconds since the epoch, when it has one
`Key`	`Option<Bytes>`	the record key, when the record has one

Topic and Partition are in the prelude; the other keys are imported from context::keys. A body that reads several fields takes ctx: &mut Context<'_, KafkaContext> instead and calls ctx.context(key) per field, as the manual form does. Reading the context also hands it to the retry position, where ToSourceTopic reads the topic. The same context carries the Position and SeekHandle keys, which move the subscription through the log.

Example

Macros
Manual

examples/kafka_context_fields.rs
/// Each `Ctx<K>` binds one field of the record; the order's customer is its record key.
#[subscriber(KafkaTopic::new("orders").group("audit"), reply)]
async fn audit(
    order: &Order,
    Ctx(topic): Ctx<Topic>,
    Ctx(partition): Ctx<Partition>,
    Ctx(offset): Ctx<Offset>,
    Ctx(timestamp_millis): Ctx<TimestampMillis>,
    Ctx(key): Ctx<Key>,
) -> AuditRecord {
    let customer = key.map(|key: Bytes| String::from_utf8_lossy(&key).into_owned());
    tracing::info!(order.id, %topic, partition, offset, "order audited");
    AuditRecord {
        order_id: order.id,
        topic: topic.to_string(),
        partition,
        offset,
        timestamp_millis,
        customer,
    }
}

examples/manual/kafka_context_fields.rs
/// The body reads each field of the record off its context by its key; the order's customer is
/// its record key.
struct Audit;

impl Handle<Order, AuditRecord, (), KafkaContext> for Audit {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, KafkaContext>,
    ) -> impl Future<Output = Result<AuditRecord, HandlerOutcome>> {
        let topic = ctx.context(Topic).to_string();
        let partition = ctx.context(Partition);
        let offset = ctx.context(Offset);
        let customer = ctx
            .context(Key)
            .map(|key| String::from_utf8_lossy(key).into_owned());
        tracing::info!(order.id, %topic, partition, offset, "order audited");
        ready(Ok(AuditRecord {
            order_id: order.id,
            topic,
            partition,
            offset,
            timestamp_millis: ctx.context(TimestampMillis),
            customer,
        }))
    }
}

fn audited() -> KafkaTopic {
    KafkaTopic::new("orders").group("audit")
}

Example​

Example