Struct KafkaPublisher 

pub struct KafkaPublisher { /* private fields */ }

A producer handle sharing the broker’s connection.

OutgoingMessage::name is the destination topic. A PARTITION_KEY_HEADER header becomes the record’s native key, so Kafka routes messages that share a key to the same partition; without it the configured partitioner picks one.

Each publish awaits the broker’s delivery report, so an Ok means the cluster accepted the record (durability then depends on the producer’s acks setting, configurable through KafkaBroker::producer_config).

transactional_id upgrades the handle to a transactional one implementing TransactionalPublisher: publishes between begin_transaction and commit become visible atomically (readers on Kafka’s default read_committed isolation see all of them or none), and abort discards them broker-side.

Obtained from KafkaBroker::publisher; usable before Broker::connect resolves the connection (publishing earlier returns KafkaError::NotConnected).

Implementations

impl KafkaPublisher

pub fn queue_timeout(self, timeout: Duration) -> Self

How long a publish may wait for space when librdkafka’s local queue is full, before failing with a queue-full error. Without it a publish waits for space indefinitely, which is the natural back-pressure behavior.

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

Upgrades to a transactional publisher fenced by id (Kafka’s transactional.id).

The id must be stable and unique per concurrent producer: Kafka uses it to fence zombies, so two live producers sharing an id abort each other. Create several publishers with distinct ids for concurrent transactional flows. The transactional producer itself is created (and its transactions initialized) on first use, from the broker’s resolved producer configuration.

Examples

use ruststream_rdkafka::KafkaBroker;

let broker = KafkaBroker::new(["localhost:9092"]);
let replies = broker.publisher().transactional_id("orders-svc-1");

pub fn transaction_timeout(self, timeout: Duration) -> Self

How long transaction control calls (init, commit, abort) may block before reporting failure. Defaults to 30 seconds; this is the call deadline handed to librdkafka, not its transaction.timeout.ms (reachable through KafkaBroker::producer_config).

Only meaningful after transactional_id.

Trait Implementations

impl Clone for KafkaPublisher

fn clone(&self) -> KafkaPublisher

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for KafkaPublisher

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

Formats the value using the given formatter.

impl Publisher for KafkaPublisher

async fn publish(&self, msg: OutgoingMessage<'_>) -> Result<(), Self::Error>

Publishes msg to the topic named by OutgoingMessage::name and awaits the delivery report. Inside an open transaction the record joins it; otherwise it goes out through the broker’s shared plain producer, transactional id or not.

Errors

Returns KafkaError::NotConnected before Broker::connect resolves the connection and KafkaError::Publish when the cluster rejects the record or the delivery times out (librdkafka’s message.timeout.ms).

Cancel safety

Not cancel safe: dropping the future may leave the record in flight, delivered or not.

type Error = KafkaError

The error type returned by publish.

impl TransactionalPublisher for KafkaPublisher

async fn begin_transaction(&self) -> Result<(), Self::Error>

Begins a Kafka transaction (creating and initializing the transactional producer on first use).

One producer runs one transaction at a time, so beginning while one is open is an error, not a queue: a second begin means two flows share one publisher, and silently merging their messages into one transaction would commit one flow’s records with the other’s. Concurrent transactional flows use distinct publishers (see TransactionalPartitions).

Errors

Returns KafkaError::InvalidOptions without a transactional_id, KafkaError::TransactionBusy when a transaction is already open on this publisher (or a clone sharing its id), KafkaError::NotConnected before Broker::connect, and KafkaError::Publish when initialization or the begin call fails.

async fn commit(&self) -> Result<(), Self::Error>

Commits the open transaction, making its records visible atomically; a no-op when none is open.

Errors

Returns KafkaError::Publish when the commit fails. librdkafka distinguishes retriable failures from ones requiring an abort; after an error the transaction’s state is unresolved, so treat the publisher as needing an abort or replacement.

async fn abort(&self) -> Result<(), Self::Error>

Aborts the open transaction, discarding its records broker-side; a no-op when none is open.

Errors

Returns KafkaError::Publish when the abort fails.