Struct KafkaTopic 

pub struct KafkaTopic { /* private fields */ }

A subscription to one Kafka topic through one consumer group.

Everything except the topic name is optional; unset options fall back to the librdkafka defaults (this crate does not impose its own). The group can also come from KafkaBroker::default_group; a subscription that ends up with no group at all is a startup error, because Kafka cannot subscribe without one.

Examples

use ruststream_rdkafka::{Assignment, Commit, KafkaTopic, StartOffset};

let topic = KafkaTopic::new("orders")
    .group("orders-svc")
    .start(StartOffset::Earliest)
    .commit(Commit::Tracked)
    .assignment(Assignment::CooperativeSticky)
    .config("fetch.min.bytes", "1024");
assert_eq!(topic.topic(), "orders");

Implementations

impl KafkaTopic

pub fn new(topic: impl Into<String>) -> Self

Describes a subscription to topic with librdkafka defaults for everything else.

pub fn pattern(pattern: impl Into<String>) -> Self

Describes a subscription to every existing topic matching pattern.

The pattern is a librdkafka topic regex and must start with ^ (that anchor is how librdkafka distinguishes a pattern from a literal name); subscribing fails with a clear error otherwise. Topics created after the group formed are picked up on the next metadata refresh.

Examples

use ruststream_rdkafka::KafkaTopic;

let orders = KafkaTopic::pattern("^orders\\..*").group("orders-svc");
assert_eq!(orders.topic(), "^orders\\..*");

pub fn and_topic(self, topic: impl Into<String>) -> Self

Adds another topic to the same subscription: one consumer, one group, several topics.

All matched topics share the handler (and therefore its payload type). Entries starting with ^ are librdkafka regex patterns, exactly as in pattern.

Examples

use ruststream_rdkafka::KafkaTopic;

let both = KafkaTopic::new("orders").and_topic("cancellations");
assert_eq!(both.topic(), "orders,cancellations");

pub fn group(self, group: impl Into<String>) -> Self

The consumer group for this subscription, overriding KafkaBroker::default_group.

pub fn start(self, start: StartOffset) -> Self

Where the group starts when it has no committed offset (see StartOffset).

pub fn commit(self, commit: Commit) -> Self

How processed deliveries are committed (see Commit).

pub fn assignment(self, assignment: Assignment) -> Self

The partition assignment strategy (see Assignment); unset means the librdkafka default (range,roundrobin).

pub fn lane_key(self, lane_key: LaneKey) -> Self

What drives keyed worker lanes for this subscription (see LaneKey); the default lanes by the source partition, Kafka’s native ordering unit.

Examples

use ruststream_rdkafka::{KafkaTopic, LaneKey};

// Opt into finer, per-record-key lanes: one tenant never processes concurrently,
// different tenants in one partition do.
let topic = KafkaTopic::new("orders")
    .group("orders-svc")
    .lane_key(LaneKey::RecordKey);

pub fn partitions(self, partitions: impl IntoIterator<Item = i32>) -> Self

Switches the subscription to manual partition assignment: the consumer assign()s exactly these partitions of the topic - no group membership, no rebalancing.

Deliveries start per start; with a group also named the consumer commits into it without joining it (so StartOffset::Committed resumes from the group’s positions), and without one commits are off and the start offset must be explicit. Does not combine with and_topic / pattern (manual assignment names exact partitions of one topic) or with Commit::Transactional.

Examples

use ruststream_rdkafka::{KafkaTopic, StartOffset};

// An inspection reader pinned to partition 0, no group side effects.
let topic = KafkaTopic::new("orders")
    .partitions([0])
    .start(StartOffset::Earliest);

pub fn retry(self, retry: Retry) -> Self

What nack(true) does on this subscription (see Retry); unset keeps Kafka’s native behavior - the offset stays unsettled and redelivers on the next fetch of the partition.

Examples

use ruststream_rdkafka::{KafkaTopic, Retry};

let topic = KafkaTopic::new("orders")
    .group("orders-svc")
    .retry(Retry::Topic("orders.retry".into()))
    .max_deliveries(5)
    .dead_letter("orders.dlq");

pub fn max_deliveries(self, max_deliveries: u32) -> Self

The poison cap: how many times a message may be delivered before nack(true) takes the drop path instead of retrying (the original delivery counts as one). Enforced by the retry policy - through RETRY_COUNT_HEADER for Retry::Topic, and through an in-session counter for Retry::SeekBack.

pub fn dead_letter(self, topic: impl Into<String>) -> Self

The dead-letter topic for the drop path: nack(false) and an exhausted retry republish the message there (stamped with the kafka-dlq-source-* headers), then settle. Without it the drop path just settles.

pub fn config(self, key: impl Into<String>, value: impl Into<String>) -> Self

Raw librdkafka consumer property passthrough for anything not surfaced as a typed option, applied last (it wins over the typed options and the broker-wide config).

pub fn topic(&self) -> &str

The subscribed name(s), joined with , when there are several (also the handler metadata name); a pattern subscription returns the pattern.

Trait Implementations

impl Clone for KafkaTopic

fn clone(&self) -> KafkaTopic

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for KafkaTopic

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

Formats the value using the given formatter.

impl SubscriptionSource<KafkaBroker> for KafkaTopic

type Subscriber = KafkaSubscriber

The subscriber type this source opens.

fn name(&self) -> &str

The name (subject / channel) this subscription binds to.

async fn subscribe( self, broker: &KafkaBroker, ) -> Result<Self::Subscriber, KafkaError>

Opens the subscription against broker. Called once, after Broker::connect.

impl SubscriptionSource<KafkaTestBroker> for KafkaTopic

Available on crate feature testing only.

type Subscriber = KafkaTestSubscriber

The subscriber type this source opens.

fn name(&self) -> &str

The name (subject / channel) this subscription binds to.

async fn subscribe( self, broker: &KafkaTestBroker, ) -> Result<Self::Subscriber, KafkaError>

Opens the subscription against broker. Called once, after Broker::connect.