Struct KafkaSubscriber 

pub struct KafkaSubscriber { /* private fields */ }

A consumer-group member on one topic, yielding KafkaMessage deliveries.

Created by subscribing a KafkaTopic descriptor (or a bare topic name) through KafkaBroker. The subscriber owns a dedicated librdkafka consumer; dropping it closes the consumer, which leaves the group and (under auto-commit) commits the final stored position. Under Commit::Tracked each in-flight delivery keeps the consumer alive, so the close happens once the last outstanding message settles or drops - do not rely on subscriber drop as an immediate group-departure barrier.

Back-pressure: polling the stream is what drives the consumer, so consuming slower simply fetches slower; librdkafka’s own fetch queue bounds (queued.max.messages.kbytes and friends, settable through KafkaTopic::config) cap local buffering.

Implementations

impl KafkaSubscriber

pub fn topic(&self) -> &str

The topic this subscriber consumes.

Trait Implementations

impl BatchSubscriber for KafkaSubscriber

fn batches( &mut self, ) -> impl Stream<Item = Result<Self::Batch, <Self as Subscriber>::Error>> + Send + '_

Streams non-empty pages natively: each waits for one delivery, then drains everything librdkafka has already fetched. There is no crate-imposed window - the page is bounded by librdkafka’s own fetch-queue limits (queued.max.messages.kbytes and friends, settable through KafkaTopic::config); wrap the source in the core Buffered adapter for an explicit size/deadline window. A consumer error inside an open page yields the page first; the error (if it persists) surfaces on the next poll.

Cancel safety

Same guarantees as Subscriber::stream: cancel safe between polls, no delivery is lost by dropping the stream.

type Batch = Vec<KafkaMessage>

Container yielded by batches. Implementations choose between Vec, custom iterators, or anything else that yields the underlying Subscriber::Message.

impl Debug for KafkaSubscriber

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

Formats the value using the given formatter.

impl Subscriber for KafkaSubscriber

fn stream( &mut self, ) -> impl Stream<Item = Result<Self::Message, Self::Error>> + Send + '_

Streams deliveries as they arrive; the stream yields an error item when the consumer fails (it does not end on its own - drop the subscriber to leave the group).

Errors librdkafka is already retrying by itself are not forwarded as stream items: today that is exactly UnknownTopicOrPartition (a subscribed topic pending creation). Such an episode surfaces as one warning when it starts - the monitoring signal to act on - with debug lines for the repeats and for the recovery, so a topic that appears late (broker auto-creation, provisioning races) recovers without flooding the dispatch error log, while a topic that never appears leaves the warning standing. Everything else is forwarded.

Cancel safety

Polling is cancel safe (the underlying recv is documented cancellation safe, so no delivery is lost by dropping the stream between polls), and the stream can be re-created by calling stream again: deliveries buffer in the consumer, not in the returned stream.

type Message = KafkaMessage

The message type yielded by this subscriber.

type Error = KafkaError

The error type yielded by the stream when delivery fails.