Skip to content

Topics and groups

A subscription is one consumer joining one consumer group on one topic. KafkaTopic is the descriptor: everything besides the topic name is optional, and unset options mean the librdkafka defaults - this crate does not impose its own.

// Everything besides the topic is optional: unset options mean the librdkafka defaults. The
// group overrides the broker-wide `default_group`; `Commit::Tracked` turns `ack` into a precise
// per-message acknowledgement backed by the stored position; `config` reaches any consumer
// property this crate does not surface as a typed option.
#[subscriber(
    KafkaTopic::new("orders")
        .group("orders-workers")
        .start(StartOffset::Earliest)
        .commit(Commit::Tracked)
        .assignment(Assignment::CooperativeSticky)
        .config("fetch.min.bytes", "1024"),
    publish("confirmations")
)]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation {
        id: order.id,
        accepted: true,
    }
}

#[ruststream::app]
fn app() -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        let confirmations = TypedPublisher::new(b.broker().publisher());
        b.include_publishing(confirm, confirmations);
        b.include(audit_partition_zero);
    })
}

Consumer groups

Kafka cannot subscribe without a group, so every subscription needs one from somewhere: KafkaTopic::group per subscription, or KafkaBroker::default_group for everything that does not name its own (including the bare-string #[subscriber("orders")] form). A subscription that ends up with no group at all is a clear startup error, not a silent default.

Start offsets

StartOffset maps to librdkafka's auto.offset.reset and applies when the group has no valid committed offset for a partition: it has never committed one, or the committed offset was deleted by retention / is out of range (which is how a long-idle group on the default latest reset skips to the end instead of reprocessing):

  • Committed (default) - leave the choice to librdkafka (its default resets to the latest offset).
  • Earliest - start from the earliest retained offset.
  • Latest - only messages published after the group formed.

Commit modes

Kafka commits one position per partition instead of settling individual messages. The Commit mode picks how ack and nack map onto that model:

  • Commit::Auto (default, the librdkafka behavior): positions are stored as messages are handed to the application and committed every auto.commit.interval.ms. ack and nack are advisory no-ops. A crash can lose the tail of processed-but-uncommitted work or skip unprocessed deliveries that were already stored.
  • Commit::Tracked: precise at-least-once. enable.auto.offset.store is switched off, and an ack advances the stored position to just below the lowest still-unsettled delivery, so out-of-order acks (concurrent worker lanes) never commit past an unprocessed message, and offset gaps the consumer never receives (transaction markers, compacted-away records) cannot block the position. Auto-commit still flushes the stored position in the background and once more when the consumer closes.
  • Commit::Transactional("pipeline-id"): exactly-once. The consumer never commits its own offsets - the EosPipeline with the matching transactional id commits them inside the producer transaction, atomically with the records the handlers publish. ack advances the shared watermark exactly like Tracked; see Exactly-once pipelines.

Negative settlement under Tracked:

  • nack(false) (drop) settles the offset so the position can move past it.
  • nack(true) (requeue) leaves the offset unsettled: the committed position stays below it, so Kafka redelivers from there when the partition is next fetched (a rebalance or a restart). The unsettled offset also blocks the watermark, keeping every later ack uncommitted until then - precise, but worth knowing when a handler nacks in a loop. Retry topics, seek-back redelivery, and dead-letter routing are planned descriptor options.

Multiple topics and patterns

One subscription can consume several topics through one consumer and one group. All matched topics share the handler, and therefore its payload type; each delivery still reports the topic it came from:

// `and_topic` adds topics to the same subscription: one consumer joins the group for both.
#[subscriber(KafkaTopic::new("orders").and_topic("cancellations").group("orders-svc"))]
async fn on_order_event(event: &OrderEvent) -> HandlerResult {
    println!("order event {}", event.id);
    HandlerResult::Ack
}

For open-ended sets, a librdkafka topic regex subscribes to every matching topic - the pattern must start with ^ (librdkafka's anchor for distinguishing patterns from literal names):

// A `^`-anchored librdkafka regex subscribes to every matching topic; topics created later are
// picked up on the next metadata refresh.
#[subscriber(KafkaTopic::pattern("^audit\\..*").group("audit-svc").start(StartOffset::Earliest))]
async fn on_audit(event: &OrderEvent) -> HandlerResult {
    println!("audit event {}", event.id);
    HandlerResult::Ack
}

The in-process test broker supports multi-topic subscriptions (exact-name routing per topic) but not patterns.

Partition assignment

KafkaTopic::assignment picks how the group balances partitions across members (librdkafka's partition.assignment.strategy): Assignment::Range, Assignment::RoundRobin, or Assignment::CooperativeSticky for incremental rebalancing where unaffected partitions keep flowing during a rebalance. Unset means the librdkafka default (range,roundrobin). Cooperative and eager strategies cannot mix within one group, and librdkafka offers no API for custom group assignors.

Manual partition assignment

KafkaTopic::partitions switches the subscription from the group protocol (subscribe) to manual assignment (assign): the consumer takes exactly the named partitions of the topic - no group membership, no rebalancing. It is the honest answer to "consume exactly these partitions": static pinning, inspection and replay readers, one-consumer-per-partition deployments.

// Manual assignment: consume exactly these partitions - no group membership, no rebalancing.
// This reader names no group, so it cannot commit and the start offset must be explicit; add
// `.group("...")` to commit positions into a group without joining it.
#[subscriber(
    KafkaTopic::new("orders")
        .partitions([0])
        .start(StartOffset::Earliest)
)]
async fn audit_partition_zero(order: &Order) -> HandlerResult {
    println!("partition 0 saw order {}", order.id);
    HandlerResult::Ack
}

A group is optional and changes only offset handling. With one named, the consumer commits into the group without joining it: Commit::Tracked stores positions exactly as on a normal subscription, and StartOffset::Committed resumes from them. Without one, commits are off and the start offset must be explicit (Earliest or Latest); Commit::Tracked and StartOffset::Committed are clear startup errors, and acks are advisory (librdkafka insists on a group.id even for assign(), so a group-less reader runs under the inert ruststream.standalone placeholder - it never joins or commits). Manual assignment does not combine with and_topic/pattern (it names exact partitions of one topic) or with Commit::Transactional, and the in-process test broker rejects it (it does not simulate partitions).

Manual assignment composes with keyed worker lanes out of the box: under the default LaneKey::Partition each assigned partition lanes independently, so partitions([0, 2, 5]) + workers(n, by_key) processes every assigned partition in order on its lane. Sizing n against the partition list is your call - fewer lanes than partitions means partitions share lanes (ordering still holds), more means idle lanes.

Keyed worker lanes

Kafka partitions by the native record key, and this crate surfaces it through IncomingMessage::partition_key (with the Partitioned capability mirroring it), so keyed lanes keep per-key ordering end to end:

// Eight lanes by record key (the opt-in): two orders for the same tenant never process
// concurrently (per-tenant ordering), while different tenants run in parallel - even inside
// one partition.
#[subscriber(
    KafkaTopic::new("orders").group("orders-workers").lane_key(LaneKey::RecordKey),
    workers(8, by_key)
)]
async fn on_order(order: &Order) -> HandlerResult {
    println!("order {} for tenant {}", order.id, order.tenant);
    HandlerResult::Ack
}

KafkaTopic::lane_key picks what drives the lanes. The default, LaneKey::Partition, lanes by the source partition - Kafka's native ordering unit, so everything one partition delivers (keyless included) processes in order on one lane, and concurrency comes from consuming several partitions:

// The default lanes by the source partition, Kafka's own ordering unit: everything one
// partition delivers (keyless audit events included) processes in order on one lane.
#[subscriber(KafkaTopic::new("audit").group("audit-workers"), workers(8, by_key))]
async fn on_audit(order: &Order) -> HandlerResult {
    println!("audit entry {} for tenant {}", order.id, order.tenant);
    HandlerResult::Ack
}

LaneKey::RecordKey opts into finer, per-record-key lanes (the first example above): deliveries sharing a record key stay ordered while different keys of one partition process concurrently. Keyless deliveries then carry no lane key and rotate across lanes, losing even their partition order.

Retries and dead-lettering

Without a policy, nack(true) keeps Kafka's native meaning: the offset stays unsettled and redelivers on the next fetch of the partition. KafkaTopic::retry gives it an immediate one:

  • Retry::Topic("orders.retry") republishes the message to the retry topic with an attempt counter riding in the kafka-retry-count header, then settles the original (republish-first: a crash between the steps duplicates, never loses).
  • Retry::SeekBack seeks the partition back and re-consumes the message in place; everything after it on that partition replays too, and the attempt count survives only within the session.
  • Retry::Drop treats nack(true) like the drop path.

max_deliveries(n) is the poison cap (the original delivery counts as one): once the next retry would exceed it, the drop path runs instead. dead_letter("orders.dlq") routes the drop path - nack(false) included - to a dead-letter topic, stamped with kafka-dlq-source-topic / -partition / -offset headers, then settles; without it the drop path just settles. Retry and dead-letter topics are your infrastructure: the crate only publishes to them. The in-process test broker does not run these policies (nack(true) re-enqueues in place there).

The usual pipeline puts the retry topic on the same subscription with and_topic, so retried copies come back to the same handler:

// `and_topic` puts the retry topic on the same subscription, so retried copies come back to
// this very handler: a fresh delivery arrives from "payments", each retry from
// "payments.retry" with the attempt count riding in the `kafka-retry-count` header. Once the
// next retry would exceed `max_deliveries`, the drop path dead-letters the message instead of
// republishing it again.
#[subscriber(
    KafkaTopic::new("payments")
        .and_topic("payments.retry")
        .group("payments-svc")
        .commit(Commit::Tracked)
        .retry(Retry::Topic("payments.retry".into()))
        .max_deliveries(5)
        .dead_letter("payments.dlq")
)]
async fn charge(payment: &Payment, State(gateway): State<PaymentGateway>) -> HandlerResult {
    if payment.amount_cents <= 0 {
        // Malformed input is not worth retrying: `drop()` takes the drop path straight to the
        // dead-letter topic.
        return HandlerResult::drop();
    }
    match gateway.charge(payment).await {
        Ok(()) => HandlerResult::Ack,
        // A transient failure: republish to "payments.retry" and settle the original, so the
        // partition keeps flowing past it.
        Err(err) => {
            eprintln!("payment {} failed: {err}; retrying", payment.id);
            HandlerResult::retry()
        }
    }
}

Retry::SeekBack trades throughput for strict partition order - nothing overtakes a failed message while it retries:

// `Retry::SeekBack` re-consumes the failed delivery in place: redelivery is immediate and the
// partition order holds (nothing overtakes the failed message), at the price of replaying
// everything after it on that partition and of the attempt count only surviving within the
// session.
#[subscriber(
    KafkaTopic::new("ledger")
        .group("ledger-svc")
        .commit(Commit::Tracked)
        .retry(Retry::SeekBack)
        .max_deliveries(3)
        .dead_letter("ledger.dlq")
)]
async fn post_entry(payment: &Payment) -> HandlerResult {
    println!("posting ledger entry for payment {}", payment.id);
    HandlerResult::Ack
}

A dead-letter consumer is an ordinary subscription; the kafka-dlq-source-* headers carry the origin of the failed delivery:

// The dead-letter consumer: dropped messages arrive stamped with the origin of the failed
// delivery in the `kafka-dlq-source-*` headers, so alerting can trace them back without
// parsing payloads.
#[subscriber(KafkaTopic::new("payments.dlq").group("payments-dlq").start(StartOffset::Earliest))]
async fn on_dead_letter(payment: &Payment, ctx: &mut Context<'_>) -> HandlerResult {
    let source = ctx
        .headers()
        .get_str(DLQ_SOURCE_TOPIC_HEADER)
        .unwrap_or("<unknown>");
    println!("payment {} dead-lettered from {source}", payment.id);
    HandlerResult::Ack
}

Batches

Batching is native: the subscriber implements the core BatchSubscriber capability directly, and a page is one delivery plus everything librdkafka has already fetched - no added waiting, no crate-imposed knobs. Page size is bounded by librdkafka's own fetch-queue limits (queued.max.messages.kbytes and friends, reachable through the raw config passthrough):

// The `batch(..)` marker wraps the usual source: batching is native - a page is one delivery
// plus everything librdkafka has already fetched, bounded by librdkafka's own fetch-queue
// limits. Returning one HandlerResult settles the whole page uniformly. `workers(4)` keeps up
// to four pages in flight at once; `by_key` does not apply to batches (a keyed policy here
// behaves like a plain pool - per-key lanes exist for single-message handlers, see the keyed
// lanes example).
#[subscriber(
    batch(KafkaTopic::new("orders").group("orders-svc").commit(Commit::Tracked)),
    workers(4)
)]
async fn handle_page(orders: &[Order]) -> HandlerResult {
    let first = orders.first().map(|order| order.id);
    println!(
        "processing a page of {} orders, first id {first:?}",
        orders.len()
    );
    HandlerResult::Ack
}

Need an explicit page window instead? The core Buffered adapter wraps any source and closes a page at max_size deliveries or max_wait after the first one (see the second handler in the example below). Batch handlers mount with include_batch:

#[ruststream::app]
fn app() -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // Kafka has no native delayed redelivery: retry_after re-publishes a delayed copy
        // through this publisher (and settles the original, so the committed position moves).
        let retries = b.broker().publisher();
        b.retry_via(retries);
        b.include_batch(handle_page);
        b.include_batch(reconcile_page);
    })
}

How batch settlement maps onto Kafka

A batch handler settles its page either uniformly (one HandlerResult for every element) or per element (Vec<HandlerResult> / Vec<Settle>, entry i settling element i):

// Per-element settlement: entry `i` settles page element `i`. This works, but read the docs on
// how it maps onto Kafka's one-position-per-partition commits: under `Commit::Tracked` the
// committed position only advances up to the first non-acked element, and `retry_after` runs
// through the runtime's deferred-republish fallback (`retry_via` below), not a native delay.
// Wrapping the source in the core `Buffered` adapter is the opt-in for an explicit
// size/deadline page window on top of the native batching.
#[subscriber(batch(
    Buffered::<KafkaTopic>::new(
        KafkaTopic::new("payments").group("payments-svc").commit(Commit::Tracked)
    )
    .max_size(50)
    .max_wait(Duration::from_millis(20))
))]
async fn reconcile_page(payments: &[Payment]) -> Vec<HandlerResult> {
    payments
        .iter()
        .map(|payment| {
            if payment.settled {
                HandlerResult::Ack
            } else {
                println!("payment {} not settled yet; retrying later", payment.id);
                HandlerResult::retry_after(Duration::from_secs(30))
            }
        })
        .collect()
}

Kafka commits one position per partition, not per message, so the outcomes map as follows (everything below assumes Commit::Tracked; under Commit::Auto every settlement is an advisory no-op and none of it applies):

  • Uniform Ack - works exactly as expected: the page settles, the position advances.
  • Per-element with Acks only - also exact: acks may even land out of order across concurrent pages, the position always advances to just below the lowest unsettled delivery.
  • Per-element with a retry() in the middle - the runtime does settle each element individually, but the committed position stops in front of the first retried element and stays there until that offset redelivers (the next fetch of the partition: a rebalance or a restart). When it does, the acked tail behind it replays too - at-least-once duplicates, not loss. Selective ack therefore works "up to the first nack" as far as the committed position is concerned; use a retry policy (retry topics) when a poison element must not hold the page hostage.
  • Per-element with retry_after(..) - Kafka has no native delayed redelivery, so the runtime's deferred-republish fallback runs: with retry_via(publisher) configured on the scope, the element settles immediately (the position moves past it) and a copy republishes to the topic's tail after the delay - ordering is not preserved, and the copy is at-most-once across the delay window (a crash before the timer fires loses it). Without retry_via the delay is dropped with a warning and the element degrades to a plain retry() hole as above.
  • A too-short result vector - the unmatched remainder of the page is retried (an extra redelivery beats a silently lost message) and the mismatch is logged.

Concurrency

workers(n) on a batch registration keeps up to n pages in flight at once; the tracked position stays correct under out-of-order acks by construction. by_key does not apply to batches - a keyed policy behaves like a plain pool of the same size. Per-key ordering is a single-message-handler feature (workers(n, by_key), see the keyed lanes example), where it composes with Kafka's native record-key partitioning end to end.

The in-process test broker batches natively the same way (a page drains what is enqueued), and the Buffered wrapper works over both brokers.

Consume errors

The subscriber stream forwards consumer errors as stream items, except the ones librdkafka is already retrying by itself - today exactly UnknownTopicOrPartition, a subscribed topic that does not exist yet. Such an episode surfaces as one warning when it starts - the monitoring signal to act on - with debug lines for the repeats and the recovery, so a topic that appears late (broker auto-creation, provisioning races) recovers on its own without flooding the dispatch error log, while a topic that never appears leaves the warning standing.

Raw configuration passthrough

KafkaTopic::config(key, value) reaches any librdkafka consumer property this crate does not surface as a typed option (fetch.min.bytes, session.timeout.ms, isolation.level, ...). It is applied last, so it wins over the typed options; overriding the keys a commit mode relies on (enable.auto.offset.store) changes that mode's behavior accordingly.