Struct KafkaConsumer 

pub struct KafkaConsumer<K, V>
where
    K: Send + Sync + 'static,
    V: Send + Sync + 'static,
{ /* private fields */ }

A client that consumes records from a Kafka cluster.

Translation notes

The concrete type lets callers dispatch statically: calling a Consumer method on a KafkaConsumer, or through generic code taking C: Consumer<K, V>, involves no boxed future. To hold it next to other consumer implementations behind one type, box it as DynConsumer.

The consumer is Send but not Sync: it can move to another task or thread, but it cannot be shared between them. This is the type-level form of Java’s KafkaConsumer.acquire() guard, which throws ConcurrentModificationException when a second thread calls into a consumer another thread is using: every Consumer method Java guards takes &mut self, and without Sync not even the methods taking &self can be reached from two threads at once. To reach the consumer from another task or thread — to call wakeup() while it waits, or to call back into it from a rebalance listener — take a ConsumerHandle with Consumer::handle, which is Send + Sync.

The consumer cannot be shared between threads:

use confluent_kafka::consumer::KafkaConsumer;

fn assert_sync<T: Sync>() {}
assert_sync::<KafkaConsumer<String, String>>();

and its getters cannot be called through a shared reference:

use std::collections::HashSet;

use confluent_kafka::common::TopicPartition;
use confluent_kafka::consumer::{Consumer, KafkaConsumer};

fn read_assignment(consumer: &KafkaConsumer<String, String>) -> HashSet<TopicPartition> {
    consumer.assignment()
}

Implementations

impl<K, V> KafkaConsumer<K, V>

where K: Send + Sync + 'static, V: Send + Sync + 'static,

pub fn new( config: ConsumerConfig, key_deserializer: impl Deserializer<K>, value_deserializer: impl Deserializer<V>, ) -> Result<Self, Error>

A consumer is instantiated by providing a set of key-value pairs as configuration, and a key and a value Deserializer.

Valid configuration strings are documented at ConsumerConfig.

Note: after creating a KafkaConsumer you must always close it to avoid resource leaks.

Arguments

• config - The consumer configs
• key_deserializer - The deserializer for key that implements Deserializer. The configure() method won’t be called in the consumer when the deserializer is passed in directly.
• value_deserializer - The deserializer for value that implements Deserializer. The configure() method won’t be called in the consumer when the deserializer is passed in directly.

Returns

The new consumer.

Errors

• Error::is_local_illegal_argument_error if group.protocol is invalid.
• Error::is_local_unsupported_operation_error if group.protocol is classic.
• Any error building the consumer from config.

Translation notes

The configuration is a built ConsumerConfig, and the deserializers are not configuration keys. Only group.protocol=consumer (KIP-848) is supported.

Trait Implementations

impl<K, V> Consumer<K, V> for KafkaConsumer<K, V>

where K: Send + Sync + 'static, V: Send + Sync + 'static,

fn assignment(&mut self) -> HashSet<TopicPartition>

Get the set of partitions currently assigned to this consumer. If subscription happened by directly assigning partitions using Self::assign then this will simply return the same partitions that were assigned. If topic subscription was used, then this will give the set of topic partitions currently assigned to the consumer (which may be none if the assignment hasn’t happened yet, or the partitions are in the process of getting reassigned).

fn subscription(&mut self) -> HashSet<String>

Get the current subscription. Will return the same topics used in the most recent call to Self::subscribe_with_topics or Self::subscribe_with_topics_listener, or an empty set if no such call has been made.

fn paused(&mut self) -> HashSet<TopicPartition>

Get the set of partitions that were previously paused by a call to Self::pause.

fn group_metadata(&mut self) -> Arc<dyn ConsumerGroupMetadata>

Return the current group metadata associated with this consumer.

fn current_lag( &mut self, topic_partition: &TopicPartition, ) -> impl Future<Output = Result<Option<i64>, Error>> + Send

Get the consumer’s current lag on the partition. Returns None if the lag is not known, for example if there is no position yet, or if the end offset is not known yet.

fn metrics(&self) -> MetricsView

Get the metrics kept by the consumer

fn subscribe_with_topics( &mut self, topics: Vec<String>, ) -> impl Future<Output = Result<(), Error>> + Send

Subscribe to the given list of topics to get dynamically assigned partitions. Topic subscriptions are not incremental. This list will replace the current assignment (if there is one). It is not possible to combine topic subscription with group management with manual partition assignment through Self::assign. If the given list of topics is empty, it is treated the same as Self::unsubscribe.

fn subscribe_with_topics_listener( &mut self, topics: Vec<String>, listener: Arc<dyn ConsumerRebalanceListener>, ) -> impl Future<Output = Result<(), Error>> + Send

Subscribe to the given list of topics to get dynamically assigned partitions. Topic subscriptions are not incremental. This list will replace the current assignment (if there is one). Note that it is not possible to combine topic subscription with group management with manual partition assignment through Self::assign. If the given list of topics is empty, it is treated the same as Self::unsubscribe.

fn subscribe_with_pattern( &mut self, pattern: SubscriptionPattern, ) -> impl Future<Output = Result<(), Error>> + Send

Subscribe to all topics matching the specified pattern, to get dynamically assigned partitions. The pattern matching will be done periodically against topics. This is only supported under the CONSUMER group protocol (see ConsumerConfig::GROUP_PROTOCOL_CONFIG)

fn subscribe_with_pattern_listener( &mut self, pattern: SubscriptionPattern, listener: Arc<dyn ConsumerRebalanceListener>, ) -> impl Future<Output = Result<(), Error>> + Send

Subscribe to all topics matching the specified pattern, to get dynamically assigned partitions. The pattern matching will be done periodically against all topics. This is only supported under the CONSUMER group protocol (see ConsumerConfig::GROUP_PROTOCOL_CONFIG).

fn assign( &mut self, partitions: Vec<TopicPartition>, ) -> impl Future<Output = Result<(), Error>> + Send

Manually assign a list of partitions to this consumer. This interface does not allow for incremental assignment and will replace the previous assignment (if there is one).

fn unsubscribe(&mut self) -> impl Future<Output = Result<(), Error>> + Send

Unsubscribe from topics currently subscribed with Self::subscribe_with_topics or Self::subscribe_with_pattern. This also clears any partitions directly assigned through Self::assign.

fn poll( &mut self, timeout: Duration, ) -> impl Future<Output = Result<ConsumerRecords<K, V>, Error>> + Send

Fetch data for the topics or partitions specified using one of the subscribe/assign APIs. It is an error to not have subscribed to any topics or partitions before polling for data.

fn commit_sync(&mut self) -> impl Future<Output = Result<(), Error>> + Send

Commit offsets returned on the last poll() for all the subscribed list of topics and partitions.

fn commit_sync_with_timeout( &mut self, timeout: Duration, ) -> impl Future<Output = Result<(), Error>> + Send

Commit offsets returned on the last poll() for all the subscribed list of topics and partitions.

fn commit_sync_with_offsets( &mut self, offsets: HashMap<TopicPartition, OffsetAndMetadata>, ) -> impl Future<Output = Result<(), Error>> + Send

Commit the specified offsets for the specified list of topics and partitions.

fn commit_sync_with_offsets_timeout( &mut self, offsets: HashMap<TopicPartition, OffsetAndMetadata>, timeout: Duration, ) -> impl Future<Output = Result<(), Error>> + Send

Commit the specified offsets for the specified list of topics and partitions.

fn commit_async(&mut self) -> impl Future<Output = Result<(), Error>> + Send

Commit offsets returned on the last Self::poll for all the subscribed list of topics and partition. Same as commitAsync(None)

fn commit_async_with_callback( &mut self, callback: Arc<dyn OffsetCommitCallback>, ) -> impl Future<Output = Result<(), Error>> + Send

Commit offsets returned on the last poll() for the subscribed list of topics and partitions.

fn commit_async_with_offsets_callback( &mut self, offsets: HashMap<TopicPartition, OffsetAndMetadata>, callback: Arc<dyn OffsetCommitCallback>, ) -> impl Future<Output = Result<(), Error>> + Send

Commit the specified offsets for the specified list of topics and partitions to Kafka.

fn seek_with_offset( &mut self, partition: TopicPartition, offset: i64, ) -> impl Future<Output = Result<(), Error>> + Send

Overrides the fetch offsets that the consumer will use on the next poll(timeout). If this API is invoked for the same partition more than once, the latest offset will be used on the next poll(). Note that you may lose data if this API is arbitrarily used in the middle of consumption, to reset the fetch offsets

fn seek_with_offset_and_metadata( &mut self, partition: TopicPartition, offset_and_metadata: OffsetAndMetadata, ) -> impl Future<Output = Result<(), Error>> + Send

Overrides the fetch offsets that the consumer will use on the next poll(timeout). If this API is invoked for the same partition more than once, the latest offset will be used on the next poll(). Note that you may lose data if this API is arbitrarily used in the middle of consumption, to reset the fetch offsets. This method allows for setting the leaderEpoch along with the desired offset.

fn seek_to_beginning( &mut self, partitions: &[TopicPartition], ) -> impl Future<Output = Result<(), Error>> + Send

Seek to the first offset for each of the given partitions. This function evaluates lazily, seeking to the first offset in all partitions only when Self::poll or Self::position are called. If no partitions are provided, seek to the first offset for all of the currently assigned partitions.

fn seek_to_end( &mut self, partitions: &[TopicPartition], ) -> impl Future<Output = Result<(), Error>> + Send

Seek to the last offset for each of the given partitions. This function evaluates lazily, seeking to the final offset in all partitions only when Self::poll or Self::position are called. If no partitions are provided, seek to the final offset for all of the currently assigned partitions.

fn position( &mut self, partition: &TopicPartition, ) -> impl Future<Output = Result<i64, Error>> + Send

Get the offset of the next record that will be fetched (if a record with that offset exists). This method may issue a remote call to the server if there is no current position for the given partition.

fn position_with_timeout( &mut self, partition: &TopicPartition, timeout: Duration, ) -> impl Future<Output = Result<i64, Error>> + Send

Get the offset of the next record that will be fetched (if a record with that offset exists). This method may issue a remote call to the server if there is no current position for the given partition.

fn committed( &mut self, partitions: &[TopicPartition], ) -> impl Future<Output = Result<HashMap<TopicPartition, Option<OffsetAndMetadata>>, Error>> + Send

Get the last committed offsets for the given partitions (whether the commit happened by this process or another). The returned offsets will be used as the position for the consumer in the event of a failure.

fn committed_with_timeout( &mut self, partitions: &[TopicPartition], timeout: Duration, ) -> impl Future<Output = Result<HashMap<TopicPartition, Option<OffsetAndMetadata>>, Error>> + Send

Get the last committed offsets for the given partitions (whether the commit happened by this process or another). The returned offsets will be used as the position for the consumer in the event of a failure.

fn partitions_for( &mut self, topic: &str, ) -> impl Future<Output = Result<Vec<PartitionInfo>, Error>> + Send

Get metadata about the partitions for a given topic. This method will issue a remote call to the server if it does not already have any metadata about the given topic.

fn partitions_for_with_timeout( &mut self, topic: &str, timeout: Duration, ) -> impl Future<Output = Result<Vec<PartitionInfo>, Error>> + Send

Get metadata about the partitions for a given topic. This method will issue a remote call to the server if it does not already have any metadata about the given topic.

fn list_topics( &mut self, ) -> impl Future<Output = Result<HashMap<String, Vec<PartitionInfo>>, Error>> + Send

Get metadata about partitions for all topics that the user is authorized to view. This method will issue a remote call to the server.

fn list_topics_with_timeout( &mut self, timeout: Duration, ) -> impl Future<Output = Result<HashMap<String, Vec<PartitionInfo>>, Error>> + Send

Get metadata about partitions for all topics that the user is authorized to view. This method will issue a remote call to the server.

fn offsets_for_times( &mut self, timestamps_to_search: HashMap<TopicPartition, i64>, ) -> impl Future<Output = Result<HashMap<TopicPartition, Option<OffsetAndTimestamp>>, Error>> + Send

Look up the offsets for the given partitions by timestamp. The returned offset for each partition is the earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition. This is a blocking call. The consumer does not have to be assigned the partitions. If the message format version in a partition is before 0.10.0, i.e. the messages do not have timestamps, None will be returned for that partition.

fn offsets_for_times_with_timeout( &mut self, timestamps_to_search: HashMap<TopicPartition, i64>, timeout: Duration, ) -> impl Future<Output = Result<HashMap<TopicPartition, Option<OffsetAndTimestamp>>, Error>> + Send

Look up the offsets for the given partitions by timestamp. The returned offset for each partition is the earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition. This is a blocking call. The consumer does not have to be assigned the partitions. If the message format version in a partition is before 0.10.0, i.e. the messages do not have timestamps, None will be returned for that partition.

fn beginning_offsets( &mut self, partitions: &[TopicPartition], ) -> impl Future<Output = Result<HashMap<TopicPartition, i64>, Error>> + Send

Get the first offset for the given partitions.

fn beginning_offsets_with_timeout( &mut self, partitions: &[TopicPartition], timeout: Duration, ) -> impl Future<Output = Result<HashMap<TopicPartition, i64>, Error>> + Send

Get the first offset for the given partitions.

fn end_offsets( &mut self, partitions: &[TopicPartition], ) -> impl Future<Output = Result<HashMap<TopicPartition, i64>, Error>> + Send

Get the end offsets for the given partitions. In the default read_uncommitted isolation level, the end offset is the high watermark (that is, the offset of the last successfully replicated message plus one). For read_committed consumers, the end offset is the last stable offset (LSO), which is the minimum of the high watermark and the smallest offset of any open transaction. Finally, if the partition has never been written to, the end offset is 0.

fn end_offsets_with_timeout( &mut self, partitions: &[TopicPartition], timeout: Duration, ) -> impl Future<Output = Result<HashMap<TopicPartition, i64>, Error>> + Send

Get the end offsets for the given partitions. In the default read_uncommitted isolation level, the end offset is the high watermark (that is, the offset of the last successfully replicated message plus one). For read_committed consumers, the end offset is the last stable offset (LSO), which is the minimum of the high watermark and the smallest offset of any open transaction. Finally, if the partition has never been written to, the end offset is 0.

fn pause( &mut self, partitions: &[TopicPartition], ) -> impl Future<Output = Result<(), Error>> + Send

Suspend fetching from the requested partitions. Future calls to Self::poll will not return any records from these partitions until they have been resumed using Self::resume. Note that this method does not affect partition subscription. In particular, it does not cause a group rebalance when automatic assignment is used.

fn resume( &mut self, partitions: &[TopicPartition], ) -> impl Future<Output = Result<(), Error>> + Send

Resume specified partitions which have been paused with Self::pause. New calls to Self::poll will return records from these partitions if there are any to be fetched. If the partitions were not previously paused, this method is a no-op.

fn enforce_rebalance( &mut self, ) -> impl Future<Output = Result<(), Error>> + Send

Alert the consumer to trigger a new rebalance by rejoining the group. This is a nonblocking call that forces the consumer to trigger a new rebalance on the next Self::poll call. Note that this API does not itself initiate the rebalance, so you must still call Self::poll. If a rebalance is already in progress this call will be a no-op. If you wish to force an additional rebalance you must complete the current one by calling poll before retrying this API.

fn enforce_rebalance_with_reason( &mut self, reason: &str, ) -> impl Future<Output = Result<(), Error>> + Send

Alert the consumer to trigger a new rebalance by rejoining the group. This is a nonblocking call that forces the consumer to trigger a new rebalance on the next Self::poll call. Note that this API does not itself initiate the rebalance, so you must still call Self::poll. If a rebalance is already in progress this call will be a no-op. If you wish to force an additional rebalance you must complete the current one by calling poll before retrying this API.

fn close(&mut self) -> impl Future<Output = Result<(), Error>> + Send

Close the consumer with default leave group behavior, waiting for up to the default timeout of 30 seconds for any needed cleanup. If auto-commit is enabled, this will commit the current offsets if possible within the default timeout. See Self::close for details. Note that Self::wakeup cannot be used to interrupt close.

fn close_with_options( &mut self, options: CloseOptions, ) -> impl Future<Output = Result<(), Error>> + Send

Close the consumer cleanly. CloseOptions allows to specify a timeout and a leave group behavior. If no timeout is specified, the default timeout of 30 seconds is used. If no leave group behavior is specified, the default leave group behavior is used.

fn wakeup(&self)

Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll. The thread which is blocking in an operation will throw WakeupError. If no thread is blocking in a method which can throw WakeupError, the next call to such a method will raise it instead.

fn handle(&self) -> ConsumerHandle

Returns a Clone + Send + Sync ConsumerHandle exposing Consumer::wakeup and the reentrant-safe consumer operations, callable from a task / thread other than the one that owns the consumer.

impl<K, V> Debug for KafkaConsumer<K, V>

where K: Send + Sync + 'static, V: Send + Sync + 'static,

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.