Struct Consumer 

pub struct Consumer { /* private fields */ }

Implementations

impl Consumer

pub async fn subscribe(&mut self, topics: Vec<String>) -> Result<()>

Subscribe to topics.

In group mode (non-empty group_id), this triggers a consumer group join to receive partition assignments from the group coordinator.

In direct mode (empty group_id), this fetches all partitions of the given topics from the cluster metadata and starts consuming them.

This method blocks until partition assignment and offset initialization are complete. In group mode, this may take up to the rebalance timeout.

pub async fn assign( &mut self, topic: impl Into<String>, partitions: Vec<i32>, ) -> Result<()>

Manually assign specific partitions to consume (direct mode only).

This is the direct-mode equivalent of subscribe(), giving you fine-grained control over which topic-partitions to consume.

This method blocks until offset initialization completes.

pub async fn seek( &self, topic: impl Into<String>, partition: i32, offset: i64, ) -> Result<()>

Seek to a specific offset for a partition (direct mode only).

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

Unsubscribe from all topics.

In direct mode, this clears all partition assignments and offsets. In group mode, this also sends a LeaveGroup request and stops the heartbeat loop.

After unsubscribe(), you can call subscribe() or assign() again to start consuming different topics.

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

Block until records are available or the channel is closed.

The first call signals the background task to start fetching (if not already started by into_stream). Subsequent calls continue reading from the prefetched buffer.

Errors

Returns [KafkaError::ConnectionClosed] if the background consumer task has terminated (e.g. due to a fatal error).

pub async fn poll_timeout( &mut self, timeout: Duration, ) -> Result<Vec<ConsumerRecord>>

Poll with a timeout.

Returns an empty Vec on timeout, or [KafkaError::ConnectionClosed] if the background consumer task has terminated.

pub async fn try_poll(&mut self) -> Result<Vec<ConsumerRecord>>

Non-blocking poll: returns immediately with whatever is buffered.

Returns an empty Vec if no data is currently buffered. Does NOT indicate a closed connection — use poll to distinguish between “no data” and “consumer terminated”.

pub async fn close(&self) -> Result<()>

Close the consumer, shutting down the background task.

pub fn offsets(&self) -> &OffsetHandle

Access offset management handle.

pub fn group(&self) -> &GroupHandle

Access group coordination handle.

pub fn is_group(&self) -> bool

Whether this consumer is in group mode.

pub fn into_stream(self) -> ConsumerStream

Convert this consumer into a streaming interface that yields individual ConsumerRecord values, one at a time.

The returned ConsumerStream shuts down the background task automatically when dropped.

Examples

let mut stream = consumer.into_stream();
while let Some(record) = stream.recv().await {
    println!("Got offset: {}", record.offset);
}