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.

This is the form a deferred retry_after copy can be published back to: one topic, and a publish to it reaches every group reading it. Several topics at once take KafkaTopics and named partitions take KafkaPartitions; neither can be reached by a single publish, so the mount site names where their copies go.

Examples

use ruststream_rdkafka::prelude::*;

#[subscriber(
    KafkaTopic::new("orders")
        .group("orders-svc")
        .start(StartOffset::Earliest)
        .commit(Commit::Tracked)
        .assignment(Assignment::CooperativeSticky)
        .config("fetch.min.bytes", "1024")
)]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(KafkaBroker::new(["localhost:9092"]), |b| {
            b.include(place);
        })
}

Implementations

impl KafkaTopic

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

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

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

Per-record-key lanes: one tenant’s orders never run concurrently, different tenants in one partition do.

use ruststream_rdkafka::prelude::*;

#[subscriber(
    KafkaTopic::new("orders")
        .group("orders-svc")
        .commit(Commit::Tracked)
        .lane_key(LaneKey::RecordKey),
    workers(8, by_key)
)]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(KafkaBroker::new(["localhost:9092"]), |b| {
            b.include(place);
        })
}

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

The exception is a property the subscription’s Commit mode owns (enable.auto.offset.store, enable.auto.commit) under Commit::Tracked or Commit::Transactional: taking it back would leave the committed position to librdkafka while every ack decided nothing, so the subscription refuses to open instead. Commit::Auto owns neither and takes both.

pub fn topic(&self) -> &str

The subscribed topic, which is also the handler metadata name.

Trait Implementations

impl Clone for KafkaTopic

fn clone(&self) -> Self

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 IntoSource for KafkaTopic

type Source = KafkaTopic

The source the constructor stores.

fn into_source(self) -> Self

Builds the source.

impl RedeliveryAddressed<ConnectedKafkaBroker> for KafkaTopic

fn redelivery_address( &self, _connected: &ConnectedKafkaBroker, ) -> impl Future<Output = Result<RedeliveryAddress, KafkaError>>

The topic itself: a record produced to it reaches every group reading it, this subscription included.

Answered from the descriptor alone, so nothing is awaited.

impl SubscriptionSource<ConnectedKafkaBroker> for KafkaTopic

type Subscriber = KafkaSubscriber

The subscriber type this source opens.

type Copies = AddressedCopies

Who publishes the copies this subscription’s retries are made of, and who names where they go: AddressedCopies, NamedCopies or BrokerMoves.

fn name(&self) -> &str

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

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

Opens the subscription against the connected broker. Called once at startup.

fn channel_bindings(&self) -> Bindings

What this subscription adds to its channel in the generated AsyncAPI document.

fn operation_bindings(&self) -> Bindings

What this subscription adds to its receive operation in the document.

fn declare_retry(self, declaration: &RetryDeclaration) -> Self

where Self: Sized,

Takes the registration’s retry declaration into the descriptor, before the subscription opens.

fn declare_retry_on( &self, connected: &C, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

Carries the registration’s declaration to the broker, at the same point and before the subscription opens.

fn message_bindings(&self) -> Bindings

What this subscription adds to the messages that arrive on it.