Struct Consumer 

pub struct Consumer<T: Transport> { /* private fields */ }

An assign-only consumer over any number of partitions.

Assignment is the caller’s: no group protocol, no rebalance, no offset commit. Where the positions live is also the caller’s problem, which is what makes this usable from a system that checkpoints offsets itself.

Implementations

impl<T: Transport> Consumer<T>

pub fn builder(transport: T) -> ConsumerBuilder<T>

A staged builder, which is the guided way in — see builder.

pub async fn new( transport: T, bootstrap: &[String], client_id: &str, isolation: IsolationLevel, ) -> Result<Self>

Connect with no assignments. Add them with Self::assign.

Errors

If no bootstrap address answers.

pub async fn for_partition( transport: T, bootstrap: &[String], client_id: &str, topic: &str, partition: i32, offset: i64, isolation: IsolationLevel, ) -> Result<Self>

Connect and assign one partition — the common case, and what the single-partition callers use.

Errors

As Self::new, plus a missing topic or partition.

pub async fn assign( &mut self, topic: &str, partition: i32, offset: i64, ) -> Result<()>

Assign another partition, starting at offset.

offset may be EARLIEST, LATEST, or an absolute offset; the first two are resolved with ListOffsets before the first fetch.

Errors

If the topic does not exist, or the broker answers with an error code.

pub async fn partition_count(&mut self, topic: &str) -> Result<i32>

How many partitions topic has, from the broker’s metadata.

This client never chooses partitions for you — there is no consumer group, so nothing assigns them behind your back — but it does have to ask the broker where they live, and the count comes back with that.

Errors

If metadata cannot be refreshed, or the topic does not exist.

pub async fn subscribe( &mut self, group_id: &str, topics: Vec<String>, assignor: Box<dyn Assignor>, reset: i64, ) -> Result<()>

Join group_id and let the group decide which partitions this consumer reads.

This is the shape most programs want, and the opposite of Self::assign: partitions arrive from the group’s leader and change when membership does. Self::poll drives the membership as a side effect, so a caller that keeps polling keeps its place in the group.

reset decides where a partition starts when the group has never committed an offset for it — Kafka’s auto.offset.reset.

Errors

If the coordinator cannot be found.

pub fn set_auto_commit(&mut self, interval: Option<Duration>)

Commit positions on a timer, without the caller asking.

Kafka’s enable.auto.commit with auto.commit.interval.ms, and the same guarantee: at least once. The commit happens at the start of a Self::poll, so what is committed is where the previous poll’s records ended — records handed to the caller and not yet committed are re-delivered after a crash. A caller that needs a record committed only once it is durably handled should commit itself, with Self::commit, after it has.

Off by default, because “at least once, silently” is a worse surprise than having to ask.

pub fn set_rebalance_listener(&mut self, listener: Box<dyn RebalanceListener>)

Be told when partitions arrive and before they are taken away.

The Kafka equivalent of ConsumerRebalanceListener. See RebalanceListener for why these are synchronous.

pub fn set_group_timeouts(&mut self, session: Duration, rebalance: Duration)

How long the coordinator waits for a heartbeat before removing this member, and how long it waits for the group to rejoin a rebalance.

Kafka’s session.timeout.ms and max.poll.interval.ms. The rebalance timeout also bounds how long a JoinGroup may be held: the coordinator keeps it until every member has rejoined, so a small value fails a stuck rebalance quickly and a large one waits patiently for slow members.

Must be set before Self::subscribe; changing it afterwards would disagree with what the group was told.

pub async fn heartbeat(&mut self) -> Result<bool>

Tell the coordinator this member is alive, without polling.

This client spawns nothing, so heartbeats ride on Self::poll. That is fine for a caller that polls in a loop, and wrong for one that spends longer than session.timeout.ms handling a batch: the coordinator removes a member it has not heard from, its partitions are given to someone else, and the slow member’s next commit is rejected.

Java hides this with a background heartbeat thread and a separate max.poll.interval.ms. The equivalent here is to call this from your own task while you work — it needs only &mut Consumer, so a caller that processes on the same executor can interleave it.

Returns whether the assignment changed, which is the same signal Self::poll acts on: true means partitions were revoked or granted and any in-flight work on the old ones should stop.

Errors

If the coordinator cannot be reached. Not being in a group is not an error — it simply does nothing.

pub async fn unsubscribe(&mut self) -> Result<()>

Leave the group, giving up every partition.

Worth calling before dropping a consumer. Without it the coordinator keeps this member until its session times out — tens of seconds during which its partitions are read by nobody, and any other member joining waits out the same delay.

Errors

If the coordinator cannot be reached. The member is forgotten locally either way: the coordinator drops it at the session timeout regardless.

pub async fn commit(&mut self) -> Result<()>

Commit the position of every partition this consumer holds.

The position is the next offset to read, which is what Kafka stores and what a restart resumes from.

Errors

If this consumer is not in a group, or the group is mid-rebalance — a commit then would write an offset for a partition that may already belong to another member.

pub fn set_metadata_max_age(&mut self, age: Duration)

How long a topic’s metadata may go unrefreshed.

Five minutes by default, matching metadata.max.age.ms. It bounds how long a partition expansion goes unnoticed — see Self::take_expansions.

pub fn take_expansions(&mut self) -> Vec<(String, i32, i32)>

Topics that have grown partitions since this was last called, as (topic, before, after). Drains what it returns.

A manual assignment is never extended for you. Java does not do it either, and it should not: the whole point of Self::assign is that the caller decides what it reads. But a caller who is never told has no way to decide, and the failure is silent — records land on partitions nobody reads, no error is raised, and Self::lag looks perfect because it only covers what is assigned.

A subscribed consumer never reports here: an expansion makes it rejoin its group instead, and the leader assigns the new partitions.

pub fn positions(&self) -> BTreeMap<TopicPartition, i64>

Where this consumer would commit to, per partition: the next offset to read, not the last one read.

For exactly-once with a group, this is the map to hand Producer::send_offsets_to_transaction together with Self::group_metadata. Read it after the records it covers have been produced, or the transaction commits offsets for output it did not write.

pub fn group_metadata(&self) -> Option<GroupMetadata>

This member’s identity and fencing token, or None if it is not in a group or not currently stable.

Hand it to a transactional producer so the coordinator can reject offsets from a member that has already been replaced. Fetch it fresh per transaction — a rebalance in between invalidates it, and that is the whole point of it.

pub fn remove(&mut self, topic: &str, partition: i32)

Stop fetching a partition.

Resets the fetch sessions: the broker’s remembered partition set no longer matches ours, and correcting it with forgotten_topics_data is more machinery than a fresh full fetch costs.

pub fn pause(&mut self, partitions: &[TopicPartition])

Stop fetching these partitions without giving them up.

The partitions stay assigned and keep their positions — this is backpressure, not a revocation, and a paused consumer must keep polling or the group will decide it is gone.

Resets the fetch sessions for the same reason Self::remove does: the broker’s remembered set no longer matches ours.

pub fn resume(&mut self, partitions: &[TopicPartition])

Fetch these partitions again, from wherever they stopped.

pub fn paused(&self) -> impl Iterator<Item = (&str, i32)>

Every partition currently paused, whether or not it is still assigned.

pub fn is_paused(&self, topic: &str, partition: i32) -> bool

Whether one partition is paused.

pub fn assignments(&self) -> impl Iterator<Item = (&str, i32)>

Every partition this consumer holds.

pub fn position_of(&self, topic: &str, partition: i32) -> Option<i64>

Where the next fetch will start for one partition.

pub fn position(&self) -> i64

Where the next fetch will start, for a consumer holding exactly one partition.

Panics

If the consumer holds anything other than one assignment — with several, “the position” is not a question with an answer.

pub fn seek_to(&mut self, topic: &str, partition: i32, offset: i64)

Seek one partition. The caller owns its offsets, so this is how a restored checkpoint is applied.

pub fn seek(&mut self, offset: i64)

Seek, for a consumer holding exactly one partition.

Panics

As Self::position.

pub fn set_incremental_fetch(&mut self, incremental: bool)

Use incremental fetch sessions (KIP-227). On by default.

Turning this off makes every fetch restate every partition, which is what the client did before sessions existed — useful if a broker or proxy mishandles them.

pub fn set_prefetch(&mut self, prefetch: bool)

Keep a fetch permanently in flight. On by default.

This is what overlaps the network with the caller’s work. Without it a fetch is issued only when Self::poll is called, so every poll pays a full round trip before it can return anything; with it the request for the next poll goes out as soon as the current one is decoded, and the caller’s processing happens while the broker is already working.

Exactly one fetch per broker is outstanding, never more: the fetch session epoch advances per accepted response, so a second in-flight request would carry an epoch the broker has not reached.

pub fn set_max_wait(&mut self, max_wait: Duration)

How long a fetch waits for data before returning empty.

pub fn connection_count(&self) -> usize

The connections this consumer holds — one per broker it fetches from, not one per partition.

pub fn metadata_leader(&self, topic: &str, partition: i32) -> Option<String>

The address the cluster map names as this partition’s leader.

pub async fn list_offset( &mut self, topic: &str, partition: i32, timestamp: i64, ) -> Result<i64>

Resolve a timestamp to an offset for one partition. EARLIEST and LATEST are the two a consumer normally wants.

Errors

If the broker answers with an error code.

pub async fn end_offsets( &mut self, partitions: &[TopicPartition], ) -> Result<BTreeMap<TopicPartition, i64>>

The offset after the last record of each partition — the log end.

Under READ_COMMITTED this is the last stable offset, so it does not run ahead of what a committed reader can see, and lag computed from it does not sit permanently at the size of an open transaction.

Errors

If no leader answers.

pub async fn beginning_offsets( &mut self, partitions: &[TopicPartition], ) -> Result<BTreeMap<TopicPartition, i64>>

The offset of the oldest record still retained in each partition.

Not zero: retention and DeleteRecords move it forward, and assuming zero is how a consumer asks for an offset the broker has deleted.

Errors

As Self::end_offsets.

pub async fn offsets_for_times( &mut self, want: &[(TopicPartition, i64)], ) -> Result<BTreeMap<TopicPartition, (i64, i64)>>

The first offset at or after each timestamp, with the timestamp of the record found.

A partition with no record at or after its timestamp is absent from the result rather than present with a sentinel, the same distinction Self::committed draws. Timestamps are milliseconds since the epoch.

Errors

As Self::end_offsets.

pub async fn lag(&mut self) -> Result<BTreeMap<TopicPartition, i64>>

How far each assigned partition is behind its log end.

A partition this consumer has not read from yet has no position, so it is absent — “unknown lag” and “zero lag” are different answers, and an alert built on the second one stays quiet through a consumer that never started.

Errors

As Self::end_offsets.

pub async fn committed( &mut self, partitions: &[TopicPartition], ) -> Result<BTreeMap<TopicPartition, i64>>

Where this consumer’s group last committed, for the partitions given.

A partition with no committed offset is absent, not zero.

Errors

If this consumer is not in a group.

impl<T: Transport> Consumer<T>

pub async fn poll(&mut self) -> Result<Vec<ConsumerRecords>>

Fetch every assigned partition, one request per broker.

Records come back grouped in the batches the broker sent, because that is how the format stores them and how the filtering works. Use ConsumerRecords::iter to walk them without caring; a key, value or header list is materialised when asked for rather than at decode time.

An empty result is normal: a fetch that waits out max_wait with no new data is not an error. Positions advance past filtered records as well as returned ones, so an all-aborted fetch makes progress rather than looping.

All four compression codecs and record headers are handled; only a pre-magic-2 batch falls back to the ordinary decoder, and then only that partition pays the old cost.

Filtering happens per batch here rather than per record, which it can because transactional, control and producer_id are batch-level in the format. That is most of why this path is cheaper.

Errors

As Self::poll.