Struct ConsumerHandle

Source

pub struct ConsumerHandle { /* private fields */ }

Expand description

A Clone + Send + Sync handle to a consumer that exposes Consumer::wakeup, callable from any task or thread, and the reentrant-safe consumer operations, callable from inside a callback the consumer is running.

No Java class counterpart — it recovers a Java capability. Java’s Consumer reference is itself a freely-shareable, thread-safe reference. Application code relies on this in two ways that a bare Rust &mut self consumer cannot express:

Cross-task wakeup() — consumer.wakeup() is called from another thread while the owning thread blocks in poll() / position() (e.g. CompletableFuture.runAsync(() -> consumer.wakeup()), org.apache.kafka.clients.consumer.PlaintextConsumerTest).
In-callback reentrancy — a ConsumerRebalanceListener calls consumer.assign/seek/pause/resume/position/committed/ beginningOffsets/commit from inside onPartitionsAssigned / onPartitionsRevoked by capturing the consumer variable in the (anonymous-inner-class) listener (PlaintextConsumerCallbackTest.java).

In Rust the consumer is owned via &mut self for the duration of a blocking call, and a consumer is not Clone, so neither pattern is expressible with a bare reference. This handle captures only the already-Arc-shared, internally-synchronized consumer state, so both patterns are expressible without unsafe. The user captures the handle into their listener struct — the Rust equivalent of Java capturing the consumer variable.

Obtain one via Consumer::handle. Cheap to clone — clones share the same underlying state.

§Operations

Sync: wakeup, assignment, paused, group_metadata, current_lag.

Async (reentrant-safe consumer ops): seek_with_offset, seek_to_beginning, seek_to_end, pause, resume, position, committed, beginning_offsets, end_offsets, offsets_for_times, partitions_for, list_topics, commit_sync, commit_sync_with_offsets, commit_async, commit_async_with_callback, commit_async_with_offsets_callback, and the _with_timeout form of each timed query.

The commits run the owning consumer’s own commit path, as a Java callback calling consumer.commitSync() / commitAsync() does: they run the consumer’s queued commit callbacks first, an async commit is one a later commit_sync or close waits for, and interceptors.onCommit fires.

§Surface: KIP-1306’s `RebalanceConsumer`, plus `wakeup`

The method set is the one KIP-1306 (RebalanceConsumer, Kafka 4.5) hands a rebalance listener, under the Rust names of the Consumer methods: every Consumer method is here with &self, except the ones a listener must not call while its rebalance is in progress — poll, close, subscribe*, unsubscribe, assign, enforce_rebalance* — and the two reads it does not carry, subscription and metrics. wakeup is the one addition: Java’s Consumer.wakeup() is callable on the consumer from any thread, while the Rust consumer’s methods take &mut self, so the handle is the only object through which that cross-thread contract is honoured. cargo xtask lint-custom (check-consumer-handle) keeps the set aligned with the trait.

§Threads: the async operations work only inside a callback

Java’s KafkaConsumer is not thread-safe: acquire() rejects a call from a thread other than the one inside the consumer with ConcurrentModificationException, and a listener may call back only because the thread running it is the one that holds the consumer. The Rust consumer turns that runtime guard into &mut self; the handle, which has no borrow of the consumer, carries the guard’s other half: an async operation is admitted only while the consumer is running a user callback on the caller’s task — a ConsumerRebalanceListener method or an OffsetCommitCallback — and otherwise returns the same ConcurrentModificationError (predicate is_local_concurrent_modification_error, message HANDLE_OUTSIDE_CALLBACK_MSG) before submitting anything. So a seek or commit_sync through the handle from a second thread while the owner is in poll() is rejected, as Java rejects it. The check is made when the operation’s future runs, so a future created inside a callback and awaited after it returned is rejected too.

What is admitted is exactly the Java use: the listener or commit callback, on the task inside the owner’s blocking call, calling back in while the owner waits for it. The background task keeps spinning during the callback, so those operations complete (consumer-threading.md §31). The scope is the consumer’s, not the caller’s: during a callback, a call from a third thread passes the check as well and is serialised by the background task like any other; the handle does not identify its caller.

wakeup and the sync reads (assignment, paused, group_metadata) are not gated: Java’s wakeup() is callable from any thread by contract, and the reads take the same short locks the owning consumer’s getters do.

§Closed consumer

Once the consumer’s close() has returned, the handle refuses it as the consumer does: the async operations return its IllegalStateError (predicate is_local_illegal_state_error) ahead of the callback check, and the sync reads panic with the same message. close() marks the consumer closed only when it returns, so a listener it runs still calls back in, as Java’s listener does from close(). wakeup stays callable.

A handle from a MockConsumer is not gated either: Java’s MockConsumer has no acquire(), and the mock handle’s commits commit into the mock from any thread as the mock’s own methods do.

§Concrete `async fn`, no `#[async_trait]`

ConsumerHandle is a concrete struct, so its async methods are concrete async fn returning an anonymous future (no Pin<Box<dyn Future>>). None of its methods are on a per-record hot path.

Implementations§

Source §

impl ConsumerHandle

Source

pub fn wakeup(&self)

Fires the consumer’s wakeup() from this handle. Equivalent to calling Consumer::wakeup on the owning consumer, but callable from any task / thread without holding a reference to the consumer.

Source

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

See Consumer::assignment.

Read from the consumer’s shared SubscriptionState.

§Returns

As Consumer::assignment.

§Panics

After the consumer’s close() has returned, with Java’s IllegalStateException message, as the consumer’s own assignment.

Source

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

See Consumer::paused.

Read from the consumer’s shared SubscriptionState; on a MockConsumer handle, the mock’s own paused set.

§Returns

As Consumer::paused.

§Panics

After the consumer’s close() has returned, with Java’s IllegalStateException message, as the consumer’s own paused.

Source

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

See Consumer::group_metadata.

Read from the consumer’s shared group-metadata slot; on a MockConsumer handle, Java’s MockConsumer.groupMetadata() sentinel values.

§Returns

As Consumer::group_metadata.

§Panics

As the consumer’s own group_metadata: after its close() has returned, with Java’s IllegalStateException message, and on a consumer without group.id, with Java’s InvalidGroupIdException message (org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer#throwIfGroupIdNotDefined).

Source

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

See Consumer::current_lag.

Submits the consumer’s CurrentLag event and waits for the background task’s answer, as the consumer’s own does.

On a MockConsumer handle, Java’s MockConsumer.currentLag over the mock’s shared end offsets and positions, answered without waiting. The handle cannot seed a missing position as the mock’s own position() does (it holds no reset offsets), so a partition without a position answers Ok(None), the lag being unknown.

§Arguments

topic_partition - as Consumer::current_lag.

§Returns

As Consumer::current_lag.

§Errors

as Consumer::current_lag;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section).

Source

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

See Consumer::seek_with_offset.

§Arguments

partition - as Consumer::seek_with_offset.
offset - as Consumer::seek_with_offset.

§Errors

as Consumer::seek_with_offset;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::seek_with_offset_and_metadata.

§Arguments

partition - as Consumer::seek_with_offset_and_metadata.
offset_and_metadata - as Consumer::seek_with_offset_and_metadata.

§Errors

as Consumer::seek_with_offset_and_metadata;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::seek_to_beginning.

§Arguments

partitions - as Consumer::seek_to_beginning.

§Errors

as Consumer::seek_to_beginning;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::seek_to_end.

§Arguments

partitions - as Consumer::seek_to_end.

§Errors

as Consumer::seek_to_end;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::pause.

§Arguments

partitions - as Consumer::pause.

§Errors

as Consumer::pause;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::resume.

§Arguments

partitions - as Consumer::resume.

§Errors

as Consumer::resume;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::position.

§Arguments

partition - as Consumer::position.

§Returns

As Consumer::position.

§Errors

as Consumer::position;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::position_with_timeout.

§Arguments

partition - as Consumer::position_with_timeout.
timeout - as Consumer::position_with_timeout.

§Returns

As Consumer::position_with_timeout.

§Errors

as Consumer::position_with_timeout;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::committed.

§Arguments

partitions - as Consumer::committed.

§Returns

As Consumer::committed.

§Errors

as Consumer::committed;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::committed_with_timeout.

§Arguments

partitions - as Consumer::committed_with_timeout.
timeout - as Consumer::committed_with_timeout.

§Returns

As Consumer::committed_with_timeout.

§Errors

as Consumer::committed_with_timeout;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::partitions_for.

§Arguments

topic - as Consumer::partitions_for.

§Returns

As Consumer::partitions_for.

§Errors

as Consumer::partitions_for;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::partitions_for_with_timeout.

§Arguments

topic - as Consumer::partitions_for_with_timeout.
timeout - as Consumer::partitions_for_with_timeout.

§Returns

As Consumer::partitions_for_with_timeout.

§Errors

as Consumer::partitions_for_with_timeout;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::list_topics.

§Returns

As Consumer::list_topics.

§Errors

as Consumer::list_topics;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::list_topics_with_timeout.

§Arguments

timeout - as Consumer::list_topics_with_timeout.

§Returns

As Consumer::list_topics_with_timeout.

§Errors

as Consumer::list_topics_with_timeout;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::beginning_offsets.

§Arguments

partitions - as Consumer::beginning_offsets.

§Returns

As Consumer::beginning_offsets.

§Errors

as Consumer::beginning_offsets;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::beginning_offsets_with_timeout.

§Arguments

partitions - as Consumer::beginning_offsets_with_timeout.
timeout - as Consumer::beginning_offsets_with_timeout.

§Returns

As Consumer::beginning_offsets_with_timeout.

§Errors

as Consumer::beginning_offsets_with_timeout;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::end_offsets.

§Arguments

partitions - as Consumer::end_offsets.

§Returns

As Consumer::end_offsets.

§Errors

as Consumer::end_offsets;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::end_offsets_with_timeout.

§Arguments

partitions - as Consumer::end_offsets_with_timeout.
timeout - as Consumer::end_offsets_with_timeout.

§Returns

As Consumer::end_offsets_with_timeout.

§Errors

as Consumer::end_offsets_with_timeout;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::offsets_for_times.

§Arguments

timestamps_to_search - as Consumer::offsets_for_times.

§Returns

As Consumer::offsets_for_times.

§Errors

as Consumer::offsets_for_times;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::offsets_for_times_with_timeout.

§Arguments

timestamps_to_search - as Consumer::offsets_for_times_with_timeout.
timeout - as Consumer::offsets_for_times_with_timeout.

§Returns

As Consumer::offsets_for_times_with_timeout.

§Errors

as Consumer::offsets_for_times_with_timeout;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section);
Error::is_local_unsupported_operation_error on a MockConsumer handle.

Source

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

See Consumer::commit_sync.

Commits the offsets the background task has consumed.

On a MockConsumer handle, commits the mock’s consumed positions into its committed offsets, as Java’s MockConsumer.commitSync().

§Errors

as Consumer::commit_sync;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section).

Source

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

See Consumer::commit_sync_with_timeout.

On a MockConsumer handle, as commit_sync: Java’s MockConsumer.commitSync(Duration) ignores the timeout.

§Arguments

timeout - as Consumer::commit_sync_with_timeout.

§Errors

as Consumer::commit_sync_with_timeout;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section).

Source

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

See Consumer::commit_sync_with_offsets.

On a MockConsumer handle, commits into the mock’s committed offsets, as Java’s MockConsumer.commitSync(Map).

§Arguments

offsets - as Consumer::commit_sync_with_offsets.

§Errors

as Consumer::commit_sync_with_offsets;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section).

Source

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

See Consumer::commit_sync_with_offsets_timeout.

On a MockConsumer handle, as commit_sync_with_offsets: Java’s MockConsumer.commitSync(Map, Duration) ignores the timeout.

§Arguments

offsets - as Consumer::commit_sync_with_offsets_timeout.
timeout - as Consumer::commit_sync_with_offsets_timeout.

§Errors

as Consumer::commit_sync_with_offsets_timeout;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section).

Source

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

See Consumer::commit_async.

Commits the offsets the background task has consumed without waiting for the result; a later commit_sync or close waits for it.

On a MockConsumer handle, commits the mock’s consumed positions into its committed offsets, as Java’s MockConsumer.commitAsync().

§Errors

as Consumer::commit_async;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section).

Source

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

See Consumer::commit_async_with_callback.

As commit_async, then callback runs on the consumer’s caller task, in commit order with the consumer’s other commit callbacks.

On a MockConsumer handle, commits the mock’s consumed positions and invokes callback inline before returning, as Java’s MockConsumer.commitAsync(OffsetCommitCallback).

§Arguments

callback - as Consumer::commit_async_with_callback.

§Errors

as Consumer::commit_async_with_callback;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section).

Source

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

See Consumer::commit_async_with_offsets_callback.

Commits offsets, then callback runs as for commit_async_with_callback.

On a MockConsumer handle, commits into the mock’s committed offsets and invokes callback inline before returning, as Java’s MockConsumer.commitAsync(Map, OffsetCommitCallback).

§Arguments

offsets - as Consumer::commit_async_with_offsets_callback.
callback - as Consumer::commit_async_with_offsets_callback.

§Errors

as Consumer::commit_async_with_offsets_callback;
Error::is_local_concurrent_modification_error outside a callback the consumer is running (see ConsumerHandle’s Threads section).