Consuming records

kafka-consume is the long-running read path: incremental fetch sessions, a streaming fetcher, and three ways of deciding which partitions you own.

[dependencies]
kafka-consume = "0.4"

Which shape do you want?

NegotiatedConsumer::subscribe probes ApiVersions for the GA ConsumerGroupHeartbeat (v1+, Kafka 4.0) — a fact already in hand on every pooled connection, so it costs no extra round trip — and picks KIP-848 when the broker serves it, the classic protocol otherwise. "Otherwise" includes 3.7–3.9 brokers, which advertise only the preview's v0 while shipping the protocol disabled; the Java client draws the same 4.0+ line. Pin the choice with ConsumerConfig::with_group_protocol(GroupProtocol::Consumer | Classic). A GroupConsumer pinned against a broker below the GA floor now fails at subscribe with UnsupportedApi, rather than raising the same error from every poll where a retry loop reads it as transient and spins.

One shape no version probe can see: a 4.x broker configured group.coordinator.rebalance.protocols=classic still advertises the GA heartbeat, because advertisement follows the coordinator rather than the config, and refuses the protocol only when the first heartbeat arrives. So Auto downgrades off the refusal itself: that first poll re-subscribes the member on the classic protocol — carrying its subscription, instance id, auto-commit choice and rebalance listener across — returns no records, and every poll after it reads from the classic path. This happens strictly before the coordinator has ever admitted the member. A refusal after that is a broker that changed underneath a live group, where re-joining under another protocol would abandon an assignment other members are waiting on, so it stays an error.

GroupConsumer and ClassicConsumer both wrap a Consumer rather than replacing it: the fetch path, the sessions and the decoding are identical, and the only thing membership changes is where the assignment comes from. That is why the manual mode is not a degraded group consumer — it is the same engine with the assignment supplied by the caller.

If what you actually want is a page of a topic rather than a stream of it, you want kafka-read instead. A scan is bounded and reports progress because a UI is drawing a page; a consumer runs until told to stop, and its interesting operations — seek, pause, resume — are about changing its mind mid-stream, which a bounded scan never does.

A manually-assigned consumer

use kafka_consume::{Consumer, ConsumerConfig, Position};

async fn example(cluster: kafka_consume::Cluster) -> kafka_consume::Result<()> {
let mut consumer = Consumer::new(cluster, ConsumerConfig::new());

consumer
    .assign(
        [("orders".to_owned(), 0), ("orders".to_owned(), 1)],
        Position::Earliest,
    )
    .await?;

loop {
    for record in consumer.poll().await? {
        println!(
            "{}-{} @{}: {:?}",
            record.topic, record.partition, record.offset, record.value
        );
    }
}
}

Consumer::connect takes bootstrap addresses if you have no Cluster yet.

assign replaces the assignment rather than adding to it, and partitions that leave it are forgotten in the next fetch — which is what stops the broker holding session state for partitions nobody is reading.

An empty poll is a normal answer, not an error. A consumer at the log end is caught up; it returns after max_wait_ms with nothing. A poll loop that treats empty as a failure is a poll loop that fails on every healthy cluster.

Records are decoded with the same tolerant decoder the scan path uses. A batch that will not decode does not stall the partition and does not end the stream: the position steps past it and polling continues. Unlike ScanEvent::Malformed, the consumer does not surface those bytes to you — Tolerant decoding covers the difference.

Changing its mind mid-stream

use kafka_consume::Consumer;
fn example(consumer: &mut Consumer) -> kafka_consume::Result<()> {
consumer.seek("orders", 0, 4_200)?;    // takes effect on the next fetch
consumer.pause("orders", 1);           // stop fetching, keep the partition
consumer.resume("orders", 1);          // continue from where it stopped

let next = consumer.position("orders", 0);   // Option<i64>
let behind = consumer.lag("orders", 0);      // Option<i64>, None until a fetch reports
Ok(())
}

A seek discards anything already buffered for that partition — a seek that still delivered the old records would not be a seek. A paused partition keeps its position and its place in the assignment, so resume does not re-resolve anything.

Offsets

A manually-assigned consumer can borrow a group's offset storage without joining the group:

use kafka_consume::{Consumer, ConsumerConfig, Position};

fn handle(record: &kafka_consume::Record) -> kafka_consume::Result<()> { Ok(()) }
async fn example(cluster: kafka_consume::Cluster) -> kafka_consume::Result<()> {
let mut consumer = Consumer::new(cluster, ConsumerConfig::new().group_id("reporting"));
consumer.assign([("orders".to_owned(), 0)], Position::Earliest).await?;

// Resume where the last run stopped, rather than where `assign` started.
consumer.seek_to_committed().await?;

for record in consumer.poll().await? {
    handle(&record)?;   // …and only then commit
}

for ((topic, partition), result) in consumer.commit().await? {
    if let Err(error) = result {
        eprintln!("{topic}-{partition}: commit failed: {error}");
    }
}

let stored = consumer.committed().await?;   // HashMap<(String, i32), CommittedOffset>
Ok(())
}

Three things about this that bite if you assume otherwise:

• A committed offset is the offset of the next record to read, not the last one handled. Storing the last record's offset re-delivers it forever. commit stores the consumer's current positions, which are already that.
• Borrowing the storage is not joining the group. The commit goes out anonymously, with an empty member id and the -1 non-member sentinel. The coordinator honours that form only while the group has no members — precisely so a detached client cannot scribble over a live group's positions. Point a standalone consumer at a group that has live members and every partition comes back UNKNOWN_MEMBER_ID. Group members commit as themselves; GroupConsumer::commit and ClassicConsumer::commit do that for you.
• The result is per partition. commit returns Vec<((String, i32), Result<()>)> — the same invariant as everywhere else in this library. An auto-commit whose result nobody checks is refused silently, so check it when it matters.

Do not read __consumer_offsets yourself. committed uses OffsetFetch; the internal topic format is not a stable interface.

Handing offsets to a transaction

For a consume-process-produce loop, the offsets should not be committed by the consumer at all — they belong inside the producer's transaction, so that a crash cannot land the records and the offsets separately. Every consumer kind exposes the two things that needs:

use kafka_consume::GroupConsumer;
async fn example(
    consumer: &mut GroupConsumer,
    producer: &kafka_produce::Producer,
) -> kafka_consume::Result<()> {
producer
    .send_offsets_to_transaction(consumer.positions(), &consumer.group_metadata()?)
    .await?;
Ok(())
}

Turn auto_commit off when you do this — the transaction owns the offsets, and an auto-commit would write them outside it. The full loop, and the four ways to get it subtly wrong, are in Producing records.

Group membership (KIP-848)

use kafka_consume::{ConsumerConfig, GroupConsumer};

fn still_running() -> bool { true }
async fn example(cluster: kafka_consume::Cluster) -> kafka_consume::Result<()> {
let mut consumer =
    GroupConsumer::subscribe(cluster, ConsumerConfig::new(), "billing", ["orders", "refunds"])
        .await?
        .auto_commit(true)                    // the default
        .instance_id("billing-pod-3");        // static membership, optional

while still_running() {
    for record in consumer.poll().await? {
        println!("{}-{} @{}", record.topic, record.partition, record.offset);
    }
}

consumer.leave().await?;
Ok(())
}

poll is the heartbeat. It beats when one is due, reconciles any new assignment, and only then reads. A member that stops polling — because it is doing slow work between batches, or because its task was descheduled — is a member the coordinator eventually evicts, and the partitions go to somebody else. Keep the loop tight and do slow work elsewhere.

How long "eventually" is, is ConsumerConfig::with_rebalance_timeout_ms: the budget a member has to complete a rebalance, which in a poll-driven client is in practice the longest a caller may spend between polls. It defaults to 300s, matching Java's max.poll.interval.ms and librdkafka's — the same number in the same wire field on both group protocols. Raise it if each batch does slow work; lower it to have the group notice a departed member sooner.

ConsumerConfig::new()
    .with_rebalance_timeout_ms(600_000)   // a slow write per batch
    .with_session_timeout_ms(45_000)      // classic protocol only

Nothing is owned until the first heartbeat comes back with an assignment, so the first poll or two returning empty is expected rather than a symptom. assignment() says what is owned right now, and member_id() is the id this client generated for itself — KIP-848 inverts the classic protocol here, where the broker issues it.

instance_id makes the member static (KIP-345): a restart inside session.timeout.ms parks the assignment rather than triggering a rebalance, which is the difference between a rolling deploy that shuffles every partition and one that does not.

That same window is how long the instance id stays claimed: a restart inside it is refused with UNRELEASED_INSTANCE_ID until the previous session lapses, which is correct on both sides and worth building an exponential reconnect backoff against rather than a fixed sleep. Under KIP-848 the length of that window is not a client setting at all — ConsumerGroupHeartbeat has no session timeout field, and the coordinator applies group.consumer.session.timeout.ms (45s by default, floored at 45s by group.consumer.min.session.timeout.ms). librdkafka refuses session.timeout.ms outright under group.protocol=consumer for the same reason. To shorten it, alter the group's own config — IncrementalAlterConfigs on a GROUP config resource, which kafka-admin exposes as ConfigResource::group(..) — after an operator has lowered the broker's floor. ConsumerConfig::with_session_timeout_ms applies to classic groups only.

leave() releases the assignment — or parks it, for a static member — after running the rebalance listener and committing. Dropping the consumer without calling it leaves the group waiting out the session timeout before it notices.

The rebalance listener

Auto-commit flushes the offsets this crate tracks. A caller that keeps its own per-partition state — a windowed aggregate, a write-behind buffer, a file handle per partition — has state the library knows nothing about, and by the time poll returns the partition is gone and another member may already own it.

So the callback fires before the revocation takes effect, while this member still owns the partitions and the broker is still waiting for the acknowledgement that gives them away:

listener.on_revoke  →  auto-commit  →  drop the partitions  →  acknowledge

use futures::future::BoxFuture;
use kafka_consume::{ConsumerConfig, GroupConsumer, RebalanceListener, Result, RevokedPartition};

struct FlushOnRevoke;

impl RebalanceListener for FlushOnRevoke {
    fn on_revoke(&mut self, revoked: Vec<RevokedPartition>) -> BoxFuture<'_, Result<()>> {
        Box::pin(async move {
            for partition in revoked {
                // `position` is what auto-commit is about to store — write your
                // own checkpoint at the same offset, not at the last record's.
                println!(
                    "flushing {}-{} at {}",
                    partition.topic, partition.partition, partition.position
                );
            }
            Ok(())
        })
    }
}

async fn example(cluster: kafka_consume::Cluster) -> Result<()> {
let consumer =
    GroupConsumer::subscribe(cluster, ConsumerConfig::new(), "billing", ["orders"])
        .await?
        .on_rebalance(FlushOnRevoke);
Ok(())
}

on_assign fires after the new partitions are owned and readable, and defaults to doing nothing — gaining a partition needs no protection.

Two properties to design around:

• on_revoke is at-least-once. Dropping a poll future must be safe, so a rebalance that has been computed but not finished is held on the consumer and retried by the next poll — which means a poll cancelled during on_revoke runs it again with the same partitions. A listener that flushes twice writes the same bytes twice; a listener that never fires loses them. Make it idempotent.
• An error from the listener does not stop the rebalance. By the time it runs, the group has already moved on and the broker is waiting for the acknowledgement. Refusing to revoke would leave this member holding partitions the group has given away — the double-ownership the revoke-then-acknowledge ordering exists to prevent. An Err is logged at warn and the rebalance proceeds.

Classic groups

Only for brokers older than 4.0, or a mixed group where a Java client is pinned to group.protocol=classic. GroupConsumer is the default on 4.x and is strictly less work.

use kafka_consume::{Assignor, ClassicConsumer, ConsumerConfig};

async fn example(cluster: kafka_consume::Cluster) -> kafka_consume::Result<()> {
let mut consumer =
    ClassicConsumer::subscribe(cluster, ConsumerConfig::new(), "billing", ["orders"])
        .await?
        .assignors([Assignor::CooperativeSticky, Assignor::Range]);

for record in consumer.poll().await? {
    println!("{}-{} @{}", record.topic, record.partition, record.offset);
}
Ok(())
}

Every member needs its own Cluster. That is not a style preference but a hard requirement of this protocol: JoinGroup blocks on the coordinator, and a Kafka broker will not read a second request from a socket until it has answered the first — so two members of one group sharing a connection deadlock, and it presents as a plain timeout with nothing in any log to explain it. GroupConsumer has no such constraint.

Because JoinGroup blocks, this is also the path where with_session_timeout_ms means something: it is a real field of the request, defaulting to Java's 45s, and it must not exceed the rebalance timeout — subscribe returns InvalidRequest rather than sending an inverted pair. The join is given its own deadline derived from the rebalance timeout, so a rebalance budget larger than the connection's request_timeout is honoured without widening the timeout every other RPC uses.

The advertised order is a vote, not a demand: the coordinator intersects every member's list, each member votes for the first of its own that survived, and the most-voted protocol wins. Advertising exactly one assignor forces the issue, at the cost of failing to join any group that does not share it — INCONSISTENT_GROUP_PROTOCOL, at join time, loudly.

Eager sticky is deliberately absent: StickyAssignor carries its state in the subscription's user_data as a struct with no schema in kafka-protocol, and hand-rolling a wire format is what this codebase does not do. cooperative-sticky has no such problem, so incremental rebalancing is available.

One protocol difference worth knowing for the listener: the classic protocol revokes eagerly, so every rebalance hands on_revoke the whole assignment rather than only the partitions that end up moving. That is what range-style rebalancing does; it is not this client rounding up.

Configuration

use kafka_consume::{ConsumerConfig, Visibility};

fn example() {
let config = ConsumerConfig::new()
    .group_id("reporting")
    .visibility(Visibility::All)
    .max_wait_ms(200);

// The rest are public fields on an owned type:
let tuned = ConsumerConfig {
    partition_max_bytes: 4 * 1024 * 1024,
    ..ConsumerConfig::new()
};
}

Note the default: a consumer is CommittedOnly, while a ScanSpec is Visibility::All. A consumer is usually a pipeline stage that should not see records a transaction abandoned; a scan is usually a human looking at what is actually in the log. read_committed does not mean the broker filters for you — it sends the records plus an AbortedTransactions list and the client does the work.

ConsumerConfig and ProducerConfig still use bare setter names (.group_id(…), not .with_group_id(…)). The rest of the workspace uses the with_ prefix, and these predate it; renaming them breaks callers, so it is a deliberate future pass rather than a trickle. See STYLE.md.

Fetch sessions, for free

Every consumer keeps a KIP-227 incremental fetch session per broker. The first request establishes the assignment and every request after it sends only what changed, which in steady state is nothing at all — so a consumer holding twelve partitions across three brokers sends three fetches per round, not twenty-four, and each carries almost no request body.

A broker that drops the session (a restart, or eviction under cache pressure) answers FETCH_SESSION_ID_NOT_FOUND or INVALID_FETCH_SESSION_EPOCH. Both are recovered by opening a new session with the full assignment, and neither is ever surfaced to you: a broker restart must not kill a consumer.

kafka-read's scan and tail deliberately keep the legacy sentinel and open no session, because they are one-shot and would otherwise leave broker state behind for a client that is not coming back.

Cancel safety

Dropping a poll future may discard a fetch that was in flight. It never advances a position for records you did not receive, so the worst case is re-fetching the same records. For a group member, a rebalance computed but not carried out is held on the consumer, so a dropped poll cannot skip the callback — the next one picks it up, still ahead of the acknowledging heartbeat.

See Cancel safety.