Skip to main content

Transactions and exactly-once

This guide publishes several records as one Kafka transaction, runs transactions under a worker pool, and makes a consume-transform-produce handler exactly-once.

A Kafka transaction makes a set of records visible together. A reader at read_committed isolation, librdkafka's default, sees all of them once the transaction commits and none of them after an abort. In RustStream a transaction is a property of the publish policy at the mount site, and the handler names only the capability it uses.

All or nothing from one handler​

ship turns a paid order into one shipment line per item. An order with an empty line is dropped, and none of its lines may reach the warehouse.

src/main.rs
#[derive(Debug, PartialEq, Deserialize)]
struct Order {
    id: u64,
    items: Vec<Item>,
}

#[derive(Debug, PartialEq, Deserialize)]
struct Item {
    sku: String,
    quantity: u32,
}

/// One shipment line per item of a paid order.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "shipment-lines")]
struct ShipmentLine {
    order_id: u64,
    sku: String,
    quantity: u32,
}

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

#[subscriber(KafkaTopic::new("paid-orders").group("shipping").commit(Commit::Tracked))]
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.quantity == 0 {
            tracing::warn!(order.id, sku = %item.sku, "order with an empty line dropped");
            return match shipments.abort().await {
                Ok(()) => HandlerOutcome::drop(),
                Err(_) => HandlerOutcome::retry(),
            };
        }
        let line = ShipmentLine {
            order_id: order.id,
            sku: item.sku.clone(),
            quantity: item.quantity,
        };
        if shipments.message(&line).publish().await.is_err() {
            shipments.abort().await.ok();
            return HandlerOutcome::retry();
        }
    }
    if shipments.commit().await.is_err() {
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, lines = order.items.len(), "order shipped");
    HandlerOutcome::ack()
}

The handler takes its slot as Out<impl TransactionalPublisher, Shipments>: it names the core's TransactionalPublisher capability and nothing of Kafka. Records published between begin_transaction and commit become visible together. abort discards them, so the first line of the order with an empty line never reaches a reader. Outside an open transaction the slot publishes like a plain one. A second begin_transaction while one is open reports KafkaError::TransactionBusy, and a commit or abort with nothing open reports KafkaError::NoTransaction.

The transaction covers the published lines and not the consumed order. Commit::Tracked commits the order's offset after the handler acks it. A crash between the two delivers the order again and ships its lines a second time. The exactly-once pipeline below closes that gap.

Transactions under a worker pool​

One Kafka producer runs one transaction at a time, so the handlers of a worker pool cannot share one transactional publisher. invoice runs four lanes and gives each source partition a transaction of its own.

src/main.rs
/// One invoice line per item of a billed order.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "invoice-items")]
struct InvoiceItem {
    order_id: u64,
    sku: String,
    quantity: u32,
}

#[subscriber(
    KafkaTopic::new("billable-orders").group("invoicing").commit(Commit::Tracked),
    workers(4, by_key)
)]
async fn invoice(
    order: &Order,
    Ctx(partition): Ctx<Partition>,
    Out(lanes): Out<impl PartitionLanes>,
) -> HandlerOutcome {
    let Ok(items) = lanes.for_partition(partition).await else {
        return HandlerOutcome::retry();
    };
    if items.begin_transaction().await.is_err() {
        return HandlerOutcome::retry();
    }
    for item in &order.items {
        let line = InvoiceItem {
            order_id: order.id,
            sku: item.sku.clone(),
            quantity: item.quantity,
        };
        if items.message(&line).publish().await.is_err() {
            items.abort().await.ok();
            return HandlerOutcome::retry();
        }
    }
    if items.commit().await.is_err() {
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, partition, "order invoiced");
    HandlerOutcome::ack()
}

workers(4, by_key) keys the lanes by partition, so one partition is handled in order on one lane. The slot is PartitionLanes: for_partition returns the transactional publisher of the delivery's partition, created on its first use. Lanes that handle different partitions run their transactions at the same time. The transactional ids follow the topic's partitions, not the number of workers, so changing workers(n) changes no id.

This scheme needs the default lane key. LaneKey::RecordKey spreads one partition across several lanes, which would then share one partition's publisher and collide on its single transaction.

The exactly-once pipeline​

enrich reads a placed order and answers with the totals the warehouse plans with. Each answer is published exactly once, even when the service crashes and the order is read again.

src/main.rs
/// A placed order with the totals the warehouse plans with.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "enriched-orders")]
struct EnrichedOrder {
    id: u64,
    lines: usize,
    units: u32,
}

#[subscriber(
    KafkaTopic::new("placed-orders")
        .group("enrichment")
        .commit(Commit::Transactional("enrichment-1".into())),
    reply,
    workers(4, by_key)
)]
async fn enrich(order: &Order) -> EnrichedOrder {
    let units = order.items.iter().map(|item| item.quantity).sum();
    tracing::info!(order.id, units, "order enriched");
    EnrichedOrder {
        id: order.id,
        lines: order.items.len(),
        units,
    }
}

The handler only returns its reply; it holds no transaction code. Commit::Transactional("enrichment-1") stops the subscription from committing offsets on its own. The pipeline of the same id commits them instead, inside the transaction that carries the replies. So the consumed position and the published records move together: either both commit, or neither does.

The mount site​

The mount site names the policy behind each slot and reply.

src/main.rs
fn service(broker: KafkaBroker) -> impl App {
    RustStream::new(AppInfo::new("fulfilment", "0.1.0")).with_broker(broker, |b| {
        // One transactional id per service instance.
        b.include(ship)
            .out(Shipments, Publish::default().transactional_id("shipping-1"))
            .build();
        // One transactional id per source partition: "invoicing-1-p0", "invoicing-1-p1", ...
        b.include(invoice)
            .out(
                DefaultSlot,
                Publish::default()
                    .transactional_id("invoicing-1")
                    .per_partition(),
            )
            .build();
        // The pipeline's id is the one the subscription's commit mode names.
        b.include(enrich)
            .out_reply(EosPublish::new("enrichment-1"))
            .transform(EosReplies);
    })
}

#[ruststream::app]
fn app() -> impl App {
    service(KafkaBroker::new(["localhost:9092"]))
}

transactional_id turns the plain policy into a transactional one. One transactional id belongs to one live producer: a second producer that starts with the same id fences the first, and the first can no longer commit. Give every running instance of the service its own id, for example from the pod's ordinal.

per_partition makes the id a base: the publishers of invoice take the ids invoicing-1-p0, invoicing-1-p1 and so on.

EosPublish::new("enrichment-1") is the pipeline, and its id is the one the subscription's commit mode names. EosReplies carries each delivery's source position onto its reply, which is how the pipeline pairs the two. Without it the first reply fails instead of leaving outside the transaction.

The commit window​

The pipeline collects replies into a window and commits it every commit_interval, 100 ms unless set. At the end of a window it waits for every delivery that published into it to settle, adds their positions and the group's metadata to the transaction, and commits. So a reply becomes visible at least one commit interval after the handler returned it.

A failed publish, a failed commit or a delivery that does not settle in time aborts the window. The consumers then seek back to the last committed offsets, and the whole window is handled again in a fresh transaction. The committed output still holds each reply once.

The pipeline limits what its handlers can return and how they read:

  • A retry() from a handler holds its window open until the transaction timeout, 30 seconds unless transaction_timeout sets another, and then the window replays.
  • retry_after does not belong in a pipeline handler: its delayed copy is published outside the window and breaks the pairing of records and offsets.
  • KafkaPartitions does not combine with Commit::Transactional, and such a subscription fails to start: the pipeline commits through the consumer group, and a reader of fixed partitions takes no part in one.