Transactions

transactional_id makes the publish policy transactional: publishes between begin and commit become visible together, or not at all.

The policy it returns is KafkaTransactionalPublish, named TransactionalPublish in the prelude. It pairs into a KafkaTransactionalPublisher, which carries the core's TransactionalPublisher capability. A handler bounds its slot Out<impl TransactionalPublisher, Marker> and calls begin_transaction, then commit or abort. The producer is created and its transactions initialized at startup, when the policy pairs, so the handler holds a fenced producer from its first delivery. Readers at Kafka's default read_committed isolation see the records of a transaction at its commit; an abort discards them on the broker. Outside an open transaction the publisher sends like a plain one.

The transactional id is the fencing unit: initializing a producer with an id fences the previous producer that held it. So the id is stable and unique per running instance, for example one per replica. transaction_timeout bounds how long begin, commit and abort may block (30 seconds by default). It is the call deadline; Kafka's own transaction.timeout.ms is a producer property, set through producer_config. queue_timeout bounds the wait for space in librdkafka's local queue.

One producer runs one transaction at a time. A second begin_transaction while one is open returns KafkaError::TransactionBusy, and a commit or abort with nothing open returns KafkaError::NoTransaction. So the handlers of a workers(n) pool do not share one transactional publisher; per_partition gives each source partition a publisher of its own. The test harness records what the handler sent through the slot, committed or not, while the topic shows only what was committed.

Example

An order ships one record per item, in one transaction. An item without a name aborts the transaction, which discards the shipments already sent.

Macros

examples/kafka_transactions.rs

#[derive(OutSlot)]
#[publishes(Shipment)]
struct Shipments;

/// One shipment per item, published in one transaction.
#[subscriber("orders")]
async fn ship(
    order: &Order,
    Out(shipments): Out<impl TransactionalPublisher, Shipments>,
) -> HandlerOutcome {
    if shipments.begin_transaction().await.is_err() {
        return HandlerOutcome::retry();
    }
    for item in &order.items {
        if item.is_empty() {
            // The shipments already sent belong to the transaction, so the abort discards them.
            if shipments.abort().await.is_err() {
                return HandlerOutcome::retry();
            }
            tracing::warn!(
                order = order.id,
                "an item has no name, the order ships nothing"
            );
            return HandlerOutcome::drop();
        }
        let shipment = Shipment {
            order_id: order.id,
            item: item.clone(),
        };
        if shipments.message(&shipment).publish().await.is_err() {
            shipments.abort().await.ok();
            return HandlerOutcome::retry();
        }
    }
    if shipments.commit().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Manual

examples/manual/kafka_transactions.rs

struct Shipments;

impl OutSlot for Shipments {
    const NAME: &'static str = "Shipments";
    type Destination = Reads;
}

impl PublishedThrough<Shipments> for Shipment {}

/// One shipment per item, published in one transaction.
struct Ship;

impl<S> Handle<Order, (), Outs<(S,)>> for Ship
where
    S: OutEntry<Shipments, Wire: TransactionalPublisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(S,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let shipments = outs.get(Shipments);
        if shipments.begin_transaction().await.is_err() {
            return Err(HandlerOutcome::retry());
        }
        for item in &order.items {
            if item.is_empty() {
                // The shipments already sent belong to the transaction, so the abort discards
                // them.
                if shipments.abort().await.is_err() {
                    return Err(HandlerOutcome::retry());
                }
                tracing::warn!(
                    order = order.id,
                    "an item has no name, the order ships nothing"
                );
                return Err(HandlerOutcome::drop());
            }
            let shipment = Shipment {
                order_id: order.id,
                item: item.clone(),
            };
            if shipments.message(&shipment).publish().await.is_err() {
                shipments.abort().await.ok();
                return Err(HandlerOutcome::retry());
            }
        }
        if shipments.commit().await.is_err() {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

Macros

examples/kafka_transactions.rs

fn app() -> impl App {
    let shipments = Publish::default()
        .transactional_id("shipments-svc-1")
        .transaction_timeout(Duration::from_secs(10));
    RustStream::new(AppInfo::new("shipments", "0.1.0")).with_broker(
        KafkaBroker::new(["localhost:9092"]).default_group("shipments-svc"),
        |b| {
            b.include(ship).out(Shipments, shipments).build();
        },
    )
}

Manual

examples/manual/kafka_transactions.rs

fn app() -> impl App {
    let shipments = Publish::default()
        .transactional_id("shipments-svc-1")
        .transaction_timeout(Duration::from_secs(10));
    RustStream::new(AppInfo::new("shipments", "0.1.0")).with_broker(
        KafkaBroker::new(["localhost:9092"]).default_group("shipments-svc"),
        |b| {
            b.include(subscriber("orders", Ship).build())
                .out(Shipments, shipments)
                .build();
        },
    )
}