Struct TransactionalPartitions 

pub struct TransactionalPartitions { /* private fields */ }

Lazily materialized transactional publishers, one per source partition.

Kafka permits one open transaction per producer and one live producer per transactional id (initializing a second fences the first), so concurrent transactional handlers need one producer each. The source partition is the natural scope: under the default LaneKey::Partition worker pool a partition’s deliveries process serially on one lane, so a publisher per partition gives every lane an independent transaction with no coordination. The id set ("{base}-p{partition}") follows the topic’s partitions rather than the worker count: changing workers(n) neither changes the ids nor weakens zombie fencing - the scheme Kafka Streams uses for its per-task producers.

Not for LaneKey::RecordKey pools: record-key lanes spread one partition across lanes, so two lanes would share a partition’s publisher and collide on its single transaction (KafkaError::TransactionBusy).

Clones share the cache, so one instance in the application state serves every handler invocation.

Examples

use ruststream_rdkafka::{KafkaBroker, TransactionalPartitions};

let broker = KafkaBroker::new(["localhost:9092"]);
let publishers = TransactionalPartitions::new(broker.publisher(), "billing-svc-1");
// In a handler: the delivery's source partition picks the publisher.
let publisher = publishers.for_partition(3); // transactional id "billing-svc-1-p3"

Implementations

impl TransactionalPartitions

pub fn new(template: KafkaPublisher, id_base: impl Into<String>) -> Self

Creates the per-partition publisher set over template (which carries the broker connection and any queue_timeout); each partition’s publisher gets the transactional id "{id_base}-p{partition}". A transactional id already set on the template is ignored.

id_base must be stable across restarts and unique per service instance - it is what scopes zombie fencing.

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

The control-call deadline (KafkaPublisher::transaction_timeout) applied to each partition’s publisher. Configure before handing the set out: publishers already materialized keep their deadline.

pub fn for_partition(&self, partition: i32) -> KafkaPublisher

The publisher owning partition’s transactional id, created on first use.

partition is the delivery’s source partition (KafkaContext’s Partition field in a handler); passing anything else still works but forfeits the serialization argument that makes the per-partition scope safe.

Panics

Panics when the internal cache mutex is poisoned, which requires a prior panic while materializing a publisher (an invariant violation, not an operational failure).

Trait Implementations

impl Clone for TransactionalPartitions

fn clone(&self) -> TransactionalPartitions

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for TransactionalPartitions

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

Formats the value using the given formatter.