per_partition

per_partition gives each source partition its own transactional publisher, so a worker pool runs independent transactions.

It turns the policy into a KafkaPartitionedPublish, named PartitionedPublish in the prelude. The transactional id becomes a base: the publisher of partition 3 is fenced by "{base}-p3". A handler bounds its slot Out<impl PartitionLanes> and reads its delivery's partition through Ctx<Partition>. PartitionLanes::for_partition returns that partition's publisher, created and initialized on first use. The live value is TransactionalPartitions, and its clones share the publishers.

The scope is the source partition because one producer runs one transaction at a time. Under the default LaneKey::Partition, a workers(n, by_key) pool processes each partition in order on one lane, so each lane has a transaction of its own. The set of ids follows the topic's partitions, not the worker count: changing n keeps the ids and their fencing. It does not combine with LaneKey::RecordKey, which spreads one partition over several lanes; two lanes would then meet on one producer and get KafkaError::TransactionBusy. What a lane publishes leaves through the partition's own publisher, so a test reads it on the topic, not on the slot's record.

Example

An invoice's lines are published in the transaction of the invoice's source partition.

Macros
Manual

examples/kafka_transactions_partitions.rs
/// Bills every line of an invoice in the transaction of the invoice's source partition.
#[subscriber(
    KafkaTopic::new("invoice-batches").commit(Commit::Tracked),
    workers(4, by_key)
)]
async fn bill(
    invoice: &Invoice,
    Ctx(partition): Ctx<Partition>,
    Out(lanes): Out<impl PartitionLanes>,
) -> HandlerOutcome {
    let Ok(lane) = lanes.for_partition(partition).await else {
        return HandlerOutcome::retry();
    };
    if lane.begin_transaction().await.is_err() {
        return HandlerOutcome::retry();
    }
    for &amount in &invoice.amounts {
        let entry = InvoiceLine {
            invoice_id: invoice.id,
            amount,
        };
        if lane.message(&entry).publish().await.is_err() {
            lane.abort().await.ok();
            return HandlerOutcome::retry();
        }
    }
    if lane.commit().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/kafka_transactions_partitions.rs
/// Bills every line of an invoice in the transaction of the invoice's source partition.
struct Bill;

impl<L> Handle<Invoice, (), Outs<(L,)>, KafkaContext> for Bill
where
    L: OutEntry<DefaultSlot, Wire: PartitionLanes>,
{
    async fn handle(
        &self,
        invoice: &Invoice,
        outs: &Outs<(L,)>,
        ctx: &mut Context<'_, KafkaContext>,
    ) -> Result<(), HandlerOutcome> {
        let partition = ctx.context(Partition);
        let Ok(lane) = outs.get(DefaultSlot).for_partition(partition).await else {
            return Err(HandlerOutcome::retry());
        };
        if lane.begin_transaction().await.is_err() {
            return Err(HandlerOutcome::retry());
        }
        for &amount in &invoice.amounts {
            let entry = InvoiceLine {
                invoice_id: invoice.id,
                amount,
            };
            if lane.message(&entry).publish().await.is_err() {
                lane.abort().await.ok();
                return Err(HandlerOutcome::retry());
            }
        }
        if lane.commit().await.is_err() {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

Macros
Manual

examples/kafka_transactions_partitions.rs
fn app(servers: &str) -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
        KafkaBroker::new([servers]).default_group("billing-svc"),
        |b| {
            // One transactional id per source partition: "billing-svc-1-p0", "billing-svc-1-p1".
            let lanes = Publish::default()
                .transactional_id("billing-svc-1")
                .per_partition();
            b.include(bill).out(DefaultSlot, lanes).build();
        },
    )
}

examples/manual/kafka_transactions_partitions.rs
fn app(servers: &str) -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
        KafkaBroker::new([servers]).default_group("billing-svc"),
        |b| {
            // One transactional id per source partition: "billing-svc-1-p0", "billing-svc-1-p1".
            let lanes = Publish::default()
                .transactional_id("billing-svc-1")
                .per_partition();
            let invoices = subscriber("invoice-batches", Bill)
                .map_source(|_| KafkaTopic::new("invoice-batches").commit(Commit::Tracked))
                .workers_by_key(nonzero!(4));
            b.include(invoices.build()).out(DefaultSlot, lanes).build();
        },
    )
}

Example​

Example