Commit

Commit decides what an ack or a nack does on Kafka, which settles a consumer group by committed position, not per message.

A consumer group stores one offset per partition: the position its next member starts reading at. A handler settles one delivery at a time, so the subscription has to turn those outcomes into that one position. The descriptor chooses how with .commit(..): KafkaTopic, KafkaTopics and KafkaPartitions all take it. Commit::Auto is the default: librdkafka stores the position as a record is handed over and commits it on a timer. Commit::Tracked moves the position only past deliveries the handlers have settled, so a crash brings back what was not finished. Commit::Transactional hands the offsets to an exactly-once pipeline, which commits them in the producer's transaction. Under Tracked an ack and a drop both settle the offset, and nothing comes back. A retry hands the delivery back to its subscription instead.

Example

Macros
Manual

examples/kafka_settlement_commit.rs
/// The group's position moves past an invoice once the handler has settled it.
#[subscriber(KafkaTopic::new("invoices").group("billing").commit(Commit::Tracked))]
async fn book(invoice: &Invoice) -> HandlerOutcome {
    if invoice.amount_cents <= 0 {
        // Nothing to book: a drop settles the offset like an ack, and nothing comes back.
        tracing::warn!(invoice.id, "invoice without an amount");
        return HandlerOutcome::drop();
    }
    tracing::info!(invoice.id, "invoice booked");
    HandlerOutcome::ack()
}

examples/manual/kafka_settlement_commit.rs
/// The group's position moves past an invoice once the handler has settled it.
struct Book;

impl Handle<Invoice, (), (), (), AppState> for Book {
    fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        _ctx: &mut Context<'_, (), AppState>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if invoice.amount_cents <= 0 {
            // Nothing to book: a drop settles the offset like an ack, and nothing comes back.
            tracing::warn!(invoice.id, "invoice without an amount");
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(invoice.id, "invoice booked");
        ready(Ok(()))
    }
}

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

Example​

Example