Struct Consumer 

pub struct Consumer { /* private fields */ }

Subscribe-style consumer handle. Construct via Consumer::builder.

Implementations

impl Consumer

pub async fn assign( &self, partitions: &[(String, i32)], ) -> Result<(), ConsumerError>

Fetch partitions, in place of the current manual assignment. Kafka’s KafkaConsumer.assign.

A partition that stays assigned keeps its position. A new partition starts at the committed offset of the group, or by auto_offset_reset. An empty slice gives up the assignment, as Kafka’s assign of an empty collection calls unsubscribe.

Errors

Returns ConsumerError::InvalidArgument for an empty topic name, and ConsumerError::IllegalState while the consumer has a subscription.

impl Consumer

pub async fn commit_sync(&self) -> Result<(), ConsumerError>

Commit the current next-offsets for every assigned partition.

This method blocks until the broker acks, for at most default_api_timeout (Kafka’s default.api.timeout.ms).

Errors

Returns an error when configuration is invalid, protocol encoding fails, or the broker rejects the commit with an error that is not retriable. A retriable error, such as NOT_COORDINATOR or a failed request, is retried. Returns ConsumerError::Timeout when the call, including any wait for a concurrent commit, the broker round trips and the retries, does not finish within default_api_timeout. When a retriable error left no time for another attempt, that error is the timeout’s cause, as in Kafka’s commitSync.

pub async fn commit_offsets_sync( &self, offsets: HashMap<(String, i32), OffsetAndMetadata>, ) -> Result<(), ConsumerError>

Commit caller-selected offsets for currently assigned partitions.

This is Kafka’s commitSync(Map<TopicPartition, OffsetAndMetadata>). The call commits each offset with its leader epoch and metadata. It does not commit other partitions and does not change the fetch positions. An offset can be past the consumed position, as in Kafka. This method blocks until the broker acks, for at most default_api_timeout (Kafka’s default.api.timeout.ms).

Errors

Ok(()) means each requested offset was broker-acknowledged while this consumer continuously owned its partition, or that ownership ended and the new owner will safely replay from the prior committed offset.

Returns an error if an offset is negative, targets an unassigned partition, or the coordinator rejects the commit. A rebalance error (ConsumerError::CommitFailed or ConsumerError::RebalanceInProgress) is not retried and commits nothing: call poll() and commit again, as Kafka’s commitSync requires. A retriable error, such as NOT_COORDINATOR or a failed request, is retried. Returns ConsumerError::Timeout when the call, including any wait for a concurrent commit, the broker round trips and the retries, does not finish within default_api_timeout. When a retriable error left no time for another attempt, that error is the timeout’s cause, as in Kafka’s commitSync.

pub fn commit_async(&self) -> Result<(), ConsumerError>

Fire-and-forget commit.

This method returns after scheduling the latest offsets for a background commit. Calls made while one is queued or running are coalesced into its snapshot or one follow-up snapshot. It does NOT wait for the broker ack. It logs errors and does not return them. Kafka’s commitAsync() with its DefaultOffsetCommitCallback.

Errors

Returns ConsumerError::InvalidGroupId without a group id.

pub fn commit_async_with_callback<F>( &self, callback: F, ) -> Result<(), ConsumerError>

where F: FnOnce(&HashMap<(String, i32), OffsetAndMetadata>, Result<(), &ConsumerError>) + Send + 'static,

Commit the positions in the background, and call callback with the result. Kafka’s commitAsync(OffsetCommitCallback).

The commit is coalesced with other asynchronous commits as Consumer::commit_async describes. The callback gets the offsets and the result of the commit that covers this call. The consumer calls it on a background task.

Errors

Returns ConsumerError::InvalidGroupId without a group id. The callback then does not run.

impl Consumer

pub async fn start( bootstrap: String, client_id: Option<String>, group_id: Option<String>, session_timeout: Time, max_poll_interval: Time, max_poll_records: usize, heartbeat_interval: Time, subscription_metadata_refresh_interval: Time, subscribe: Vec<String>, subscribe_pattern: Option<TopicPattern>, exclude_internal_topics: bool, subscribe_regex: Option<String>, group_protocol: GroupProtocol, group_remote_assignor: Option<String>, group_instance_id: Option<String>, auto_offset_reset: AutoOffsetReset, isolation_level: IsolationLevel, assignors: Vec<Assignor>, fetch_min: i32, fetch_max: i32, fetch_partition_max: i32, fetch_max_wait: Time, metadata_max_age: Time, default_api_timeout: Time, request_timeout: Time, dns_timeout: Time, send_buffer: Option<ByteSize>, receive_buffer: Option<ByteSize>, dispatch_queue_capacity: usize, frame_max: ByteSize, metadata_recovery_strategy: MetadataRecoveryStrategy, metadata_recovery_rebootstrap_trigger: Time, leave_group_timeout: Time, client_rack: Option<String>, security: Option<ClientSecurity>, client_dns_lookup: ClientDnsLookup, reconnect_backoff: Time, reconnect_backoff_max: Time, connections_max_idle: Time, socket_connection_setup_timeout: Time, socket_connection_setup_timeout_max: Time, rebalance_listener: Option<Box<dyn ConsumerRebalanceListener>>, retry_backoff_max: Time, retry_policy: Option<ConsumerRetryPolicy>, enable_auto_commit: Option<bool>, auto_commit_interval: Time, allow_auto_create_topics: bool, enable_metrics_push: bool, ) -> Result<Self, ConsumerError>

Build a Consumer subscribed to the given topics.

This function validates the configuration eagerly and fails fast before any network I/O. It then calls the internal start_once operation with a per-attempt timeout. An attempt can stall on a lost wakeup during cold-boot group-join contention, or return a transient error. The function then drops the timed-out future, which cancels its in-flight connections, and starts a fresh attempt. A genuine misconfiguration or a persistent error surfaces immediately.

With enable_auto_commit on (the default, as in Kafka), the consumer commits its positions as Kafka’s enable.auto.commit does: from poll each auto_commit_interval (default 5 s), before each JoinGroup, and in close. The commit before a JoinGroup includes only records that the application received before its latest poll. Set enable_auto_commit(false) to commit manually, for example in a transactional consume-process-produce loop. The consumer then commits nothing that the application does not ask for.

max_poll_interval (default 5 min) is Kafka’s max.poll.interval.ms. When the application does not call poll for longer, a dynamic member sends LeaveGroup and a static member stops its heartbeats. The next poll joins the group again. The consumer also sends the value as the JoinGroup rebalance timeout. The consumer starts a rebalance that the coordinator asks for only in poll, so the assignment does not change for the eager protocol while the application processes the records of the last poll.

Errors

Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.

pub fn builder() -> ConsumerStartBuilder

Build a Consumer subscribed to the given topics.

This function validates the configuration eagerly and fails fast before any network I/O. It then calls the internal start_once operation with a per-attempt timeout. An attempt can stall on a lost wakeup during cold-boot group-join contention, or return a transient error. The function then drops the timed-out future, which cancels its in-flight connections, and starts a fresh attempt. A genuine misconfiguration or a persistent error surfaces immediately.

With enable_auto_commit on (the default, as in Kafka), the consumer commits its positions as Kafka’s enable.auto.commit does: from poll each auto_commit_interval (default 5 s), before each JoinGroup, and in close. The commit before a JoinGroup includes only records that the application received before its latest poll. Set enable_auto_commit(false) to commit manually, for example in a transactional consume-process-produce loop. The consumer then commits nothing that the application does not ask for.

max_poll_interval (default 5 min) is Kafka’s max.poll.interval.ms. When the application does not call poll for longer, a dynamic member sends LeaveGroup and a static member stops its heartbeats. The next poll joins the group again. The consumer also sends the value as the JoinGroup rebalance timeout. The consumer starts a rebalance that the coordinator asks for only in poll, so the assignment does not change for the eager protocol while the application processes the records of the last poll.

Errors

Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.

impl Consumer

pub fn group_id(&self) -> &str

The consumer’s group id.

pub fn member_id(&self) -> String

The current member id. It is empty while the member is not in the group, and it changes when the member joins the group again from scratch.

pub fn generation_id(&self) -> i32

The current group generation. The coordinator task keeps it live across rejoins through the shared Arc<AtomicI32>.

pub fn group_metadata(&self) -> ConsumerGroupMetadata

KIP-447 group metadata to hand to a transactional producer’s send_offsets_to_transaction.

The generation id is the coordinator’s live generation, which the shared Arc<AtomicI32> keeps current across rejoins.

pub async fn assignment(&self) -> Vec<(String, i32)>

Snapshot of currently assigned (topic, partition) pairs.

pub async fn at_log_end(&self) -> bool

Whether every assigned partition has consumed through the last end offset reported by a successful fetch.

pub async fn close(self) -> Result<(), ConsumerError>

Close the consumer with GroupMembershipOperation::Default: commit with auto commit on, stop the coordinator task, and send LeaveGroup for a dynamic member. A static member stays in the group. See close_with.

Errors

Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.

pub async fn client_instance_id( &self, timeout: Time, ) -> Result<Option<Uuid>, ConsumerError>

The client instance id that the broker assigned for metrics push (KIP-714), as Kafka’s KafkaConsumer.clientInstanceId returns it.

The call waits up to timeout for the first GetTelemetrySubscriptions response. It returns Ok(None) when none has come by then, as Kafka returns null. A zero timeout does not wait.

Errors

Returns ClientError::TelemetryDisabled when enable_metrics_push is false, and ClientError::InvalidArgument for a negative timeout, as Kafka throws an IllegalStateException and an IllegalArgumentException.

pub async fn close_with( self, options: CloseOptions, ) -> Result<(), ConsumerError>

Close the consumer, and leave the group or stay in it as options.group_membership_operation says. Kafka’s KafkaConsumer.close(CloseOptions).

options.timeout (default 30 s) bounds the auto commit (at most request_timeout) and then the close of the fetch sessions, as Kafka’s ClassicKafkaConsumer.close bounds the coordinator close and the fetcher close. The consumer always sends the LeaveGroup and waits for it for at most leave_group_timeout: Kafka’s maybeLeaveGroup transmits the request (pollNoWakeup) before awaitPendingRequests checks the close timer. The rebalance listener call before the LeaveGroup is not bounded. A negative timeout gives ConsumerError::InvalidArgument; the consumer then stops without a commit and without LeaveGroup.

The coordinator itself sends the best-effort LeaveGroup as the last thing it does on shutdown. See coordinator::run. It uses its live member_id, which can differ from the one captured at build time, because a from-scratch rejoin (UNKNOWN_MEMBER_ID) replaces it. The cancel and join are prompt, because the coordinator races its in-tick RPCs against the shutdown token.

Errors

Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.

impl Consumer

pub fn wakeup_handle(&self) -> WakeupHandle

A handle that wakes the poll of this consumer from another task. Kafka’s KafkaConsumer.wakeup is thread safe; poll takes &mut self here, so another task needs its own handle.

pub fn wakeup(&self)

Make the next poll return ConsumerError::Wakeup. Kafka’s KafkaConsumer.wakeup.

pub fn enforce_rebalance( &self, reason: Option<&str>, ) -> Result<(), ConsumerError>

Ask the group to rebalance. The next poll sends JoinGroup with reason, or with Kafka’s default reason rebalance enforced by user when reason is None or empty. Kafka’s KafkaConsumer.enforceRebalance.

Errors

Returns ConsumerError::InvalidGroupId without a group id, and ConsumerError::IllegalState for a consumer without group membership (Kafka: Tried to force a rebalance but consumer does not have a group.). A member of the consumer group protocol logs a warning and does nothing, as Kafka’s AsyncKafkaConsumer does.

impl Consumer

pub async fn pause( &self, partitions: &[(String, i32)], ) -> Result<(), ConsumerError>

Stop fetching from partitions. Kafka’s KafkaConsumer.pause.

poll does not return records of a paused partition and does not fetch it. The consumer stays in the group. Records that poll fetched before the pause come after resume. A rebalance that keeps the partition keeps the pause.

Errors

Returns ConsumerError::NoCurrentAssignment if the consumer does not own a partition. The call then pauses no partition.

pub async fn resume( &self, partitions: &[(String, i32)], ) -> Result<(), ConsumerError>

Fetch from partitions again. Kafka’s KafkaConsumer.resume.

Errors

Returns ConsumerError::NoCurrentAssignment if the consumer does not own a partition. The call then resumes no partition.

pub async fn paused(&self) -> HashSet<(String, i32)>

The paused partitions that the consumer owns. Kafka’s KafkaConsumer.paused.

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

The offset of the next record that poll fetches for (topic, partition). Kafka’s KafkaConsumer.position.

When the partition waits for an offset reset or for a leader epoch validation, the call sends ListOffsets or OffsetForLeaderEpoch and waits for the position, for at most default_api_timeout.

Errors

Returns ConsumerError::NoCurrentAssignment if the consumer does not own the partition, ConsumerError::Timeout if the position is not known before default_api_timeout, or the error of a request.

pub async fn committed( &self, partitions: &[(String, i32)], ) -> Result<HashMap<(String, i32), Option<OffsetAndMetadata>>, ConsumerError>

The committed offsets of partitions in the group. Kafka’s KafkaConsumer.committed.

A partition without a committed offset maps to None. The partitions do not need to be assigned to this consumer.

Errors

Returns the error of the OffsetFetch request after the retries that Kafka’s consumer does, for at most default_api_timeout.

impl Consumer

pub async fn poll( &mut self, timeout: Time, ) -> Result<Vec<ConsumerRecord>, ConsumerError>

Returns the records from every v2 batch that the broker returned per assigned partition, or an empty vec on timeout.

Under read_committed isolation, this method filters out control batches and records that belong to aborted transactions on the client side, with the response’s aborted_transactions list. The broker returns verbatim bytes.

poll returns at most max_poll_records records. It keeps the rest of a fetch for the next call, which then sends no Fetch.

A rebalance that the coordinator asks for starts only when the application calls poll. The internal coordinator task then runs the join and mutates the live assigned snapshot in place. For the eager protocol, poll waits for that join up to timeout and fetches nothing before it completes. poll also resets the max_poll_interval timer.

Errors

Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.

impl Consumer

pub async fn beginning_offsets( &self, partitions: &[(String, i32)], ) -> Result<HashMap<(String, i32), i64>, ConsumerError>

The first offset of each partition. Kafka’s KafkaConsumer.beginningOffsets.

A partition without an offset is not in the result.

Errors

Returns ConsumerError::TopicAuthorizationFailed, ConsumerError::Timeout when some offsets are not known before default_api_timeout, or a request error that is not retriable.

pub async fn end_offsets( &self, partitions: &[(String, i32)], ) -> Result<HashMap<(String, i32), i64>, ConsumerError>

The end offset of each partition: the high watermark, or the last stable offset with read_committed. Kafka’s KafkaConsumer.endOffsets.

A partition without an offset is not in the result.

Errors

Returns ConsumerError::TopicAuthorizationFailed, ConsumerError::Timeout when some offsets are not known before default_api_timeout, or a request error that is not retriable.

pub async fn offsets_for_times( &self, timestamps: &HashMap<(String, i32), i64>, ) -> Result<HashMap<(String, i32), Option<OffsetAndTimestamp>>, ConsumerError>

The first offset of each partition whose timestamp is at or after the given timestamp. Kafka’s KafkaConsumer.offsetsForTimes.

A partition without such a record maps to None.

Errors

Returns ConsumerError::InvalidArgument for a negative timestamp, ConsumerError::TopicAuthorizationFailed, ConsumerError::Timeout when some offsets are not known before default_api_timeout, or a request error that is not retriable.

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

The number of records between the position of an assigned partition and its end offset from the last fetch. Kafka’s KafkaConsumer.currentLag.

None when the position or the end offset is not known.

Errors

Returns ConsumerError::NoCurrentAssignment if the consumer does not own the partition.

pub async fn partitions_for( &self, topic: &str, ) -> Result<Vec<PartitionInfo>, ConsumerError>

The partitions of topic. Kafka’s KafkaConsumer.partitionsFor.

An unknown topic gives an empty list. With allow_auto_create_topics, the Metadata request lets the broker create the topic.

Errors

Returns ConsumerError::TopicAuthorizationFailed, ConsumerError::InvalidTopic, ConsumerError::Timeout when a retriable error lasts past default_api_timeout, or a request error.

pub async fn list_topics( &self, ) -> Result<HashMap<String, Vec<PartitionInfo>>, ConsumerError>

The partitions of every topic of the cluster. Kafka’s KafkaConsumer.listTopics.

Errors

Returns ConsumerError::TopicAuthorizationFailed, ConsumerError::Timeout when a retriable error lasts past default_api_timeout, or a request error.

impl Consumer

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

Set the next offset that poll fetches for (topic, partition).

offset is the next offset to read, that is the last consumed offset plus 1, the same value as a committed group offset. Pass 0 to read a partition from the beginning. The consumer does not validate the position with OffsetForLeaderEpoch, as Kafka’s seek(TopicPartition, long) does not.

Errors

Returns ConsumerError::InvalidOffset if offset is negative, and ConsumerError::NoCurrentAssignment if the consumer does not own the partition.

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

Set the next offset that poll fetches for (topic, partition), with the leader epoch of the record before offset.

This is Kafka’s seek(TopicPartition, OffsetAndMetadata) with a leader epoch. When the consumer knows the current leader epoch of the partition, it validates the position with OffsetForLeaderEpoch before the next fetch, as Kafka’s SubscriptionState.seekUnvalidated does. A truncated log then resets the position by auto.offset.reset.

Errors

Returns ConsumerError::InvalidOffset if offset is negative, and ConsumerError::NoCurrentAssignment if the consumer does not own the partition.

pub async fn seek_to_beginning( &self, partitions: &[(String, i32)], ) -> Result<(), ConsumerError>

Reset the position of partitions to the log start. Kafka’s KafkaConsumer.seekToBeginning.

An empty slice resets every assigned partition. The next poll or position sends ListOffsets with timestamp -2 and fetches from the offset that it returns.

Errors

Returns ConsumerError::NoCurrentAssignment if the consumer does not own a partition. The call then resets no partition.

pub async fn seek_to_end( &self, partitions: &[(String, i32)], ) -> Result<(), ConsumerError>

Reset the position of partitions to the log end. Kafka’s KafkaConsumer.seekToEnd.

An empty slice resets every assigned partition. The next poll or position sends ListOffsets with timestamp -1. With read_committed, that is the last stable offset.

Errors

Returns ConsumerError::NoCurrentAssignment if the consumer does not own a partition. The call then resets no partition.

impl Consumer

pub async fn subscribe<I, S>(&mut self, topics: I) -> Result<(), ConsumerError>

where I: IntoIterator<Item = S>, S: Into<String>,

Subscribe to topics, in place of the current subscription. Kafka’s KafkaConsumer.subscribe(Collection).

The next poll joins the group again with the new topics. An empty list is unsubscribe.

Errors

Returns ConsumerError::InvalidArgument for an empty topic name, and the error of the listener call of unsubscribe.

pub fn subscribe_pattern( &mut self, pattern: TopicPattern, ) -> Result<(), ConsumerError>

Subscribe to the topics that pattern matches, in place of the current subscription. Kafka’s KafkaConsumer.subscribe(Pattern).

The coordinator task matches the pattern against all topics of the cluster at once and each subscription_metadata_refresh_interval, and the next poll joins the group again when the matched topics change.

Errors

Returns ConsumerError::InvalidGroupId without a group id, and ConsumerError::IllegalState for a consumer without group membership.

pub fn subscribe_regex(&mut self, regex: &str) -> Result<(), ConsumerError>

Subscribe to the topics that the regular expression regex matches, in place of the current subscription. Kafka’s KafkaConsumer.subscribe(SubscriptionPattern) (KIP-848).

The group coordinator matches the expression with RE2J and assigns the partitions of the topics that match, so the consumer sends no topic names and matches nothing itself. The expression must follow the RE2 syntax; a broker that cannot compile it answers with INVALID_REGULAR_EXPRESSION, which poll returns.

Errors

Returns ConsumerError::InvalidArgument for an empty expression, ConsumerError::InvalidGroupId without a group id, and ConsumerError::IllegalState for a consumer with a manual assignment or with the classic group protocol.

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

Give up the subscription and the assigned partitions, and leave the group. Kafka’s KafkaConsumer.unsubscribe.

The rebalance listener gets the owned partitions first, as in close. Then the coordinator task sends LeaveGroup with the reason the consumer unsubscribed from all topics. A poll without a subscription returns ConsumerError::NotSubscribed.

Errors

Returns ConsumerError::RebalanceListenerFailed when the listener callback fails. The consumer unsubscribes also then.

pub fn subscription(&self) -> Vec<String>

The subscribed topics. For a pattern subscription, the topics that matched the last metadata. Kafka’s KafkaConsumer.subscription.

Trait Implementations

impl Drop for Consumer

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.