Skip to main content

Commits and delivery guarantees

This guide chooses what ack and nack commit on Kafka: librdkafka's own commits, precise at-least-once with Commit::Tracked, and batches settled element by element.

Kafka does not settle single messages. A consumer group commits a position per partition, and everything below it counts as read. A handler still settles each delivery, so the subscription's Commit mode decides how those settlements become a position.

ModeackdropretryAfter a crash
Auto (default)nothingnothingreports unsupported, the record does not come backthe tail of handled records is read again, or records stored but not handled are skipped
Trackedsettles the offsetsettles the offsetthe record comes backevery record not settled is read again
Transactionalsettles the offset, committed with the pipeline's transactionthe same as ackstalls the window until the transaction deadlinethe open window is read again

The example is a payouts service: payouts must not be lost, page views may be. Transactions and exactly-once covers the third mode.

At-least-once with Commit::Tracked​

pay_out sends a payout through the bank and answers with a confirmation. When the bank does not answer, the payout comes back and is tried again.

src/main.rs
#[derive(Debug, PartialEq, Deserialize)]
struct Payout {
    id: u64,
    amount_cents: u64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "payouts.done")]
struct Paid {
    payout_id: u64,
}

#[subscriber(KafkaTopic::new("payouts").commit(Commit::Tracked), reply)]
async fn pay_out(payout: &Payout, State(bank): State<Arc<Bank>>) -> Result<Paid, HandlerOutcome> {
    if payout.amount_cents == 0 {
        return Err(HandlerOutcome::drop());
    }
    if bank.0.call().await.is_err() {
        tracing::warn!(payout.id, "bank unavailable, payout comes back");
        return Err(HandlerOutcome::retry());
    }
    tracing::info!(payout.id, payout.amount_cents, "payout sent");
    Ok(Paid {
        payout_id: payout.id,
    })
}

Under Commit::Tracked an ack moves the committed position up to the lowest delivery that is still open. Acks from concurrent workers arrive out of order, and the position never passes a payout that is still in flight. retry hands the delivery back to its subscription, which delivers it again before anything it fetches next. Until then its offset stays open, so the position does not move past it. A process that stops first gets the payout back from Kafka when it starts again.

So a payout is never lost, and it may be handled twice: once before a crash and once after. The bank call has to accept a repeated payout, for example by its id. A retry without an attempt cap comes back forever; Retries and dead-letter topics adds the cap.

librdkafka's own commits​

count_view counts page views in an analytics store. A lost view does not matter here, so the subscription keeps the default mode.

src/main.rs
#[derive(Debug, PartialEq, Deserialize)]
struct PageView {
    page: String,
}

#[subscriber(KafkaTopic::new("page-views").group("analytics"))]
async fn count_view(view: &PageView, State(counters): State<Arc<Counters>>) -> HandlerOutcome {
    if counters.0.call().await.is_err() {
        // Under `Commit::Auto` nothing brings the view back; analytics accepts the loss.
        return HandlerOutcome::retry();
    }
    tracing::debug!(page = %view.page, "page view counted");
    HandlerOutcome::ack()
}

Under Commit::Auto librdkafka stores the position when it hands the record to the application and commits it on a timer. ack and drop change nothing. retry reports itself unsupported, and the view does not come back. This mode costs the least, and a crash can lose handled records or skip records that were stored but not handled.

Batches​

reconcile takes statement lines in batches of up to 500 and settles each line on its own.

src/main.rs
#[derive(Debug, PartialEq, Deserialize)]
struct StatementLine {
    payout_id: u64,
    amount_cents: u64,
}

#[subscriber(KafkaTopic::new("statements").group("reconciliation").commit(Commit::Tracked))]
async fn reconcile(lines: &[StatementLine]) -> Vec<HandlerOutcome> {
    lines
        .iter()
        .map(|line| {
            if line.amount_cents == 0 {
                tracing::warn!(line.payout_id, "empty statement line dropped");
                return HandlerOutcome::drop();
            }
            HandlerOutcome::ack()
        })
        .collect()
}

A batch is one delivery plus the records librdkafka has already fetched, with no extra waiting, so it is often smaller than the size the mount site names. The returned list settles the batch element by element; in the manual form Ok(()) acks every element and Err(outcomes) settles each one. Under Commit::Tracked an element that asks for a retry comes back alone, and the committed position stops in front of it until it settles. A process that stops first reads the acked elements behind it again: duplicates, not loss.

How much librdkafka keeps queued is a consumer property, set with KafkaTopic::config. When the queue is full, librdkafka pauses fetching for fetch.queue.backoff.ms (1000 ms by default), which slows large records behind a slow handler. .config("fetch.queue.backoff.ms", "10") shortens the pause.

The producer side​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = Clients> {
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("payouts")
        .producer_config("enable.idempotence", "true")
        .flush_timeout(Duration::from_secs(10));
    RustStream::new(AppInfo::new("payouts", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(broker, |b| {
            b.include(pay_out)
                .out_reply(Publish::default().queue_timeout(Duration::from_secs(5)));
            b.include(count_view);
            b.include(reconcile.batch(nonzero!(500usize)));
        })
}

At-least-once from end to end needs the producer side too. enable.idempotence makes librdkafka's own resends of a record land once; producer_config sets it and every other producer property, acks included. Every publish waits for the cluster's delivery report, so a publish that returned Ok was accepted by Kafka.

A publish waits for space while librdkafka's local queue is full, which slows the handler down to what the cluster takes. queue_timeout bounds that wait and turns it into an error. At shutdown the broker flushes the records still in flight, for up to flush_timeout (30 seconds by default). ClosedKafkaBroker::unflushed_records counts the records that did not make it.

The tests in the example file run the app in process and replace the bank and the analytics store with clients that fail once. The in-process Kafka commits as a cluster does, so the payout comes back under Commit::Tracked and the page view does not under Commit::Auto.