Struct MockConsumer 

pub struct MockConsumer<K, V> { /* private fields */ }

A mock of the Consumer interface you can use for testing code that uses Kafka. This class is not threadsafe. However, you can use the Self::schedule_poll_task method to write multithreaded tests where a driver thread waits for Consumer::poll to be called by a background thread and then can safely perform operations during a callback.

Translation notes

The mock is Send but not Sync: another task reaches it through Consumer::handle.

use confluent_kafka::consumer::MockConsumer;

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

Implementations

impl<K, V> MockConsumer<K, V>

pub fn new(offset_reset_strategy: &str) -> Result<Self, Error>

A mock consumer is instantiated by providing ConsumerConfig::AUTO_OFFSET_RESET_CONFIG value as the input.

Arguments

• offset_reset_strategy - the offset reset strategy to use

Returns

The new mock consumer.

Errors

• Error::is_local_illegal_argument_error if offset_reset_strategy is not a valid auto.offset.reset value

pub fn add_record(&mut self, record: ConsumerRecord<K, V>) -> Result<(), Error>

Adds a record to be returned when Consumer::poll is called.

Arguments

• record - the record to add

Errors

• Error::is_local_illegal_state_error if the partition is not assigned to the consumer

pub fn update_beginning_offsets( &mut self, offsets: HashMap<TopicPartition, i64>, )

Updates the beginning offsets for the specified partitions.

Arguments

• offsets - the beginning offsets to set

pub fn update_end_offsets(&mut self, offsets: HashMap<TopicPartition, i64>)

Updates the end offsets for the specified partitions.

Arguments

• offsets - the end offsets to set

pub fn update_duration_offsets(&mut self, offsets: HashMap<TopicPartition, i64>)

Updates the duration-based reset offsets for the specified partitions.

Arguments

• offsets - the duration offsets to set

pub fn update_partitions( &mut self, topic: &str, partitions: Vec<PartitionInfo>, ) -> Result<(), Error>

Updates the partition information for the specified topic.

Arguments

• topic - the topic to update
• partitions - the partition information

Errors

• Error::is_local_illegal_state_error if the consumer is closed

pub fn set_poll_error(&mut self, error: Option<Error>)

Sets an exception to throw when Consumer::poll is called.

Arguments

• error - the exception to throw; None clears a pending one

pub fn set_offsets_error(&mut self, error: Option<Error>)

Sets an exception to throw when offset-related methods are called.

Arguments

• error - the exception to throw; None clears a pending one

pub fn set_max_poll_records( &mut self, max_poll_records: i64, ) -> Result<(), Error>

Sets the maximum number of records returned in a single call to Consumer::poll.

Arguments

• max_poll_records - the max.poll.records.

Errors

• Error::is_local_illegal_argument_error if max_poll_records is less than 1

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

where K: Send, V: Send,

Simulate a rebalance event.

Arguments

• new_assignment - the new assignment

Errors

• The error the listener returns.
• Error::is_local_illegal_argument_error if a partition is not of a subscribed topic

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

where K: Send, V: Send,

Simulates a partition loss event. Calls ConsumerRebalanceListener::on_partitions_lost for the specified partitions and removes them from the current assignment. Unlike Self::rebalance, which calls ConsumerRebalanceListener::on_partitions_revoked, this method models the case where the consumer loses partitions without a graceful revoke.

Only records belonging to the lost partitions are cleared; records for retained partitions are unaffected.

Arguments

• partitions_lost - the partitions to lose; all must be currently assigned

Errors

• Error::is_local_illegal_state_error if any partition is not currently assigned
• The error the listener returns; the assignment is then left unchanged.

pub fn schedule_poll_task( &mut self, task: Box<dyn FnOnce(&mut MockConsumer<K, V>) + Send>, )

Schedule a task to be executed during a poll(). One enqueued task will be executed per Consumer::poll invocation. You can use this repeatedly to mock out multiple responses to poll invocations.

Arguments

• task - the task to be executed

Translation notes

The task is called with the mock, &mut MockConsumer<K, V>, which a Rust closure cannot capture.

pub fn schedule_nop_poll_task(&mut self)

Schedules a no-op task to be executed during a poll invocation.

pub fn closed(&self) -> bool

Return whether the consumer has been closed.

Returns

true if closed, false otherwise

pub fn should_rebalance(&self) -> bool

Return whether an enforceRebalance request is pending.

Returns

true if rebalance is pending, false otherwise

pub fn reset_should_rebalance(&mut self)

Resets the rebalance flag set by Consumer::enforce_rebalance.

pub fn last_poll_timeout(&self) -> Option<Duration>

Returns the timeout used in the last Consumer::poll call.

Returns

the last poll timeout, or None if poll has not been called

Trait Implementations

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

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

fn metrics(&self) -> MetricsView

Returns an empty view: the mock keeps no metrics.

Panics

Panics with "This consumer has already been closed." on a closed mock, where Java’s ensureNotClosed() throws IllegalStateException: calling it after close() is a programming error.

async fn current_lag( &mut self, topic_partition: &TopicPartition, ) -> Result<Option<i64>, Error>

Returns the end offset set by MockConsumer::update_end_offsets minus the position, or 0 when no end offset was set.

Arguments

• topic_partition - The partition to get the lag for.

Errors

• The error of Consumer::position (a closed mock, an unassigned partition).

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.

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

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.

async fn subscribe_with_topics_listener( &mut self, topics: Vec<String>, listener: Arc<dyn ConsumerRebalanceListener>, ) -> Result<(), Error>

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.

async fn subscribe_with_pattern( &mut self, pattern: SubscriptionPattern, ) -> Result<(), Error>

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)

async fn subscribe_with_pattern_listener( &mut self, pattern: SubscriptionPattern, listener: Arc<dyn ConsumerRebalanceListener>, ) -> Result<(), Error>

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).

async fn assign(&mut self, partitions: Vec<TopicPartition>) -> Result<(), Error>

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).

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

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.

async fn poll( &mut self, timeout: Duration, ) -> Result<ConsumerRecords<K, V>, Error>

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.

async fn commit_sync(&mut self) -> Result<(), Error>

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

async fn commit_sync_with_timeout( &mut self, _timeout: Duration, ) -> Result<(), Error>

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

async fn commit_sync_with_offsets( &mut self, offsets: HashMap<TopicPartition, OffsetAndMetadata>, ) -> Result<(), Error>

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

async fn commit_sync_with_offsets_timeout( &mut self, offsets: HashMap<TopicPartition, OffsetAndMetadata>, _timeout: Duration, ) -> Result<(), Error>

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

async fn commit_async(&mut self) -> Result<(), Error>

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

async fn commit_async_with_callback( &mut self, callback: Arc<dyn OffsetCommitCallback>, ) -> Result<(), Error>

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

async fn commit_async_with_offsets_callback( &mut self, offsets: HashMap<TopicPartition, OffsetAndMetadata>, callback: Arc<dyn OffsetCommitCallback>, ) -> Result<(), Error>

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

async fn seek_with_offset( &mut self, partition: TopicPartition, offset: i64, ) -> Result<(), Error>

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

async fn seek_with_offset_and_metadata( &mut self, partition: TopicPartition, offset_and_metadata: OffsetAndMetadata, ) -> Result<(), Error>

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.

async fn seek_to_beginning( &mut self, partitions: &[TopicPartition], ) -> Result<(), Error>

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.

async fn seek_to_end( &mut self, partitions: &[TopicPartition], ) -> Result<(), Error>

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.

async fn position(&mut self, partition: &TopicPartition) -> Result<i64, Error>

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.

async fn position_with_timeout( &mut self, partition: &TopicPartition, _timeout: Duration, ) -> Result<i64, Error>

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.

async fn committed( &mut self, partitions: &[TopicPartition], ) -> Result<HashMap<TopicPartition, Option<OffsetAndMetadata>>, Error>

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.

async fn committed_with_timeout( &mut self, partitions: &[TopicPartition], _timeout: Duration, ) -> Result<HashMap<TopicPartition, Option<OffsetAndMetadata>>, Error>

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.

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

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.

async fn partitions_for_with_timeout( &mut self, topic: &str, _timeout: Duration, ) -> Result<Vec<PartitionInfo>, Error>

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.

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

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

async fn list_topics_with_timeout( &mut self, _timeout: Duration, ) -> Result<HashMap<String, Vec<PartitionInfo>>, Error>

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

async fn offsets_for_times( &mut self, _timestamps_to_search: HashMap<TopicPartition, i64>, ) -> Result<HashMap<TopicPartition, Option<OffsetAndTimestamp>>, Error>

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.

async fn offsets_for_times_with_timeout( &mut self, timestamps_to_search: HashMap<TopicPartition, i64>, _timeout: Duration, ) -> Result<HashMap<TopicPartition, Option<OffsetAndTimestamp>>, Error>

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.

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

Get the first offset for the given partitions.

async fn beginning_offsets_with_timeout( &mut self, partitions: &[TopicPartition], _timeout: Duration, ) -> Result<HashMap<TopicPartition, i64>, Error>

Get the first offset for the given partitions.

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

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.

async fn end_offsets_with_timeout( &mut self, partitions: &[TopicPartition], _timeout: Duration, ) -> Result<HashMap<TopicPartition, i64>, Error>

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.

async fn pause(&mut self, partitions: &[TopicPartition]) -> Result<(), Error>

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.

async fn resume(&mut self, partitions: &[TopicPartition]) -> Result<(), Error>

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.

async fn enforce_rebalance(&mut self) -> Result<(), Error>

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.

async fn enforce_rebalance_with_reason( &mut self, _reason: &str, ) -> Result<(), Error>

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.

async fn close(&mut self) -> Result<(), Error>

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.

async fn close_with_options( &mut self, _options: CloseOptions, ) -> Result<(), Error>

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 MockConsumer<K, V>

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

Formats the value using the given formatter.

impl<K, V> Drop for MockConsumer<K, V>

Rust-only: releases the rebalance listener. A listener that captured Consumer::handle holds the shared SubscriptionState through the handle, and that state stores the listener, so without this the cycle would outlive the mock; Java’s garbage collector frees the same graph. The release waits for the drop rather than close(), because Java’s rebalance and losePartitions still run the listener on a closed mock (org.apache.kafka.clients.consumer.MockConsumer#rebalance and org.apache.kafka.clients.consumer.MockConsumer#losePartitions make no ensureNotClosed() check).

fn drop(&mut self)

Executes the destructor for this type.