Skip to content

Publishing

The outgoing message name is the destination topic: KafkaBroker::publisher() hands out one shared producer handle, and each publish awaits the cluster's delivery report, so an Ok means Kafka accepted the record.

Record keys

The partition-key header becomes the record's native key on publish (and is not duplicated as a wire header); consuming through this crate surfaces it back under the same name. Kafka routes every message for one key to one partition, which is what keeps per-key ordering:

use ruststream::{Headers, OutgoingMessage, Publisher};
use ruststream_rdkafka::{
    KafkaBroker, KafkaError, KafkaPublisher, PARTITION_HEADER, PARTITION_KEY_HEADER,
};

// The partition-key header becomes the record's native key on publish, so Kafka itself routes
// every message for one tenant to one partition (and therefore to one consumer, in order).
async fn publish_keyed(
    publisher: &KafkaPublisher,
    id: u64,
    tenant: &str,
) -> Result<(), KafkaError> {
    let mut headers = Headers::new();
    headers.insert(PARTITION_KEY_HEADER, tenant.to_owned());
    let payload = format!(r#"{{"id":{id},"tenant":"{tenant}"}}"#);
    publisher
        .publish(OutgoingMessage::new("orders", payload.as_bytes()).with_headers(headers))
        .await
}

Without the header, the configured partitioner picks a partition (the librdkafka default is a hash for keyed records and round-robin-ish distribution for keyless ones).

Explicit partitions and round-robin distribution

The partition header (kafka-partition, an ASCII decimal) pins a record to an exact partition: the publisher consumes the header (it never hits the wire) and sets the record's partition explicitly. Precedence on publish: an explicit partition wins over the record key, which wins over the configured partitioner. A malformed value fails the publish with a clear error, and a partition that does not exist fails delivery - no silent fallback either way:

// The partition header pins the record to an exact partition: the publisher consumes the
// header (it never hits the wire) and targets the partition explicitly, winning over the
// partitioner and the record key. The partition must exist, or the publish fails.
async fn publish_pinned(
    publisher: &KafkaPublisher,
    id: u64,
    partition: i32,
) -> Result<(), KafkaError> {
    let mut headers = Headers::new();
    headers.insert(PARTITION_HEADER, partition.to_string());
    let payload = format!(r#"{{"id":{id},"tenant":"pinned"}}"#);
    publisher
        .publish(OutgoingMessage::new("orders", payload.as_bytes()).with_headers(headers))
        .await
}

On top of it, RoundRobin distributes publishing-handler replies evenly: librdkafka has no round-robin partitioner (only the random/consistent/hash families), and keyless distribution may batch-stick to one partition, which turns long, near-constant per-message processing into one hot consumer and idle peers. The transform stamps every keyless, unpinned reply with the next partition of the cycle:

// Every reply targets the next partition of the cycle (0, 1, ..., 7, 0, ...), one
// message each. Replies that already carry an explicit partition or a record key are
// left alone - keys exist for ordering, and the transform must not break placement
// the handler chose. The count is explicit and must match the destination topic.
let work_items =
    TypedPublisher::new(b.broker().publisher()).transform(RoundRobin::partitions(8));
b.include_publishing(plan, work_items);

The count is explicit and must match the destination topic: a smaller count just leaves the tail partitions idle, a larger one fails publishes to the missing partitions. The header mechanism also serves any other placement policy through a user-written PublishTransform.

Delivery guarantees

Durability is the producer's acks setting, and everything else librdkafka offers (enable.idempotence, message.timeout.ms, compression, ...) is one producer_config away:

  • KafkaBroker::producer_config(key, value) - producer-only properties.
  • KafkaBroker::config(key, value) - client-wide properties (consumers and the producer).

For at-least-once end to end, pair an idempotent producer (producer_config("enable.idempotence", "true")) with Commit::Tracked on the consuming side (see Topics and groups).

Transactions

broker.publisher().transactional_id("orders-svc-1") upgrades the handle to the core TransactionalPublisher capability: 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. The id fences zombies, so it must be stable and unique per concurrent producer - create several publishers with distinct ids for concurrent transactional flows. Outside an open transaction the handle publishes like a plain one, and calling the transaction methods without an id is a clear error. The transactional producer is created and initialized on first use from the broker's resolved producer configuration; transaction_timeout bounds the control calls, while Kafka's own transaction.timeout.ms is one producer_config away. Tying consumed offsets into the producer transaction (full consume-transform-produce exactly-once) is the exactly-once pipeline below.

The usual shape is a use-case object in the application state that owns the transactional publisher and runs one atomic fan-out per call:

// The application state wires the use-case object once at startup; `#[derive(FromRef)]` makes
// each field injectable into handlers as `State<FieldType>`.
#[derive(Clone, FromRef)]
struct AppState {
    shipments: Shipments,
    invoices: Invoices,
}

#[derive(Clone)]
struct Shipments {
    publisher: KafkaPublisher,
}

impl Shipments {
    /// Publishes one shipment command per item, all-or-nothing: `commit` makes the whole batch
    /// visible atomically to `read_committed` readers, and any failure aborts so shipments are
    /// never half-visible.
    async fn dispatch(&self, order: &Order) -> Result<(), KafkaError> {
        self.publisher.begin_transaction().await?;
        for item in &order.items {
            let command = ItemShipment {
                order_id: order.id,
                item: item.clone(),
            };
            let payload = JsonCodec.encode(&command).expect("serializable");
            let outgoing = OutgoingMessage::new("shipments", payload.as_ref());
            if let Err(err) = self.publisher.publish(outgoing).await {
                self.publisher.abort().await.ok();
                return Err(err);
            }
        }
        self.publisher.commit().await
    }
}

Handlers receive it through State and settle by the outcome - an abort left nothing visible, so a retry redelivers and reruns the whole fan-out:

#[subscriber("orders")]
async fn ship(order: &Order, State(shipments): State<Shipments>) -> HandlerResult {
    if shipments.dispatch(order).await.is_err() {
        // Nothing became visible; ask for redelivery and try the whole fan-out again.
        return HandlerResult::retry();
    }
    HandlerResult::Ack
}

The id is picked where the publisher is created, one per concurrent producer:

// The transactional id must be stable and unique per concurrent producer: it is what
// fences a zombie instance. One id per service replica (pod ordinal, instance id) is the
// usual scheme.
let shipments = Shipments {
    publisher: broker.publisher().transactional_id("shipments-svc-1"),
};

Transaction scopes and worker pools

Two Kafka facts shape everything here: one producer runs one transaction at a time, and one transactional id belongs to one live producer (initializing a second fences the first). A workers(n, by_key) pool therefore cannot share a single transactional publisher - a second begin_transaction while one is open is a TransactionBusy error, deliberately: silently merging two lanes' messages into one transaction would commit one flow's records with the other's.

The scope that composes with a worker pool is the source partition. Under the default LaneKey::Partition lanes a partition's deliveries process serially on one lane, so TransactionalPartitions - a publisher per partition, ids "{base}-p{partition}" - gives every lane an independent transaction with no coordination. The id set follows the topic's partitions rather than the worker count, so changing workers(n) neither changes the ids nor weakens zombie fencing (the same scheme Kafka Streams uses for its per-task producers):

// Concurrent transactional handlers: one producer runs one transaction at a time, so a worker
// pool cannot share one publisher (a second `begin_transaction` is a `TransactionBusy` error,
// not a silent merge). Under the default partition lanes a partition processes serially on
// one lane, so a publisher per source partition gives every lane its own independent
// transaction - and the id set follows the topic's partitions, not the worker count, so
// zombie fencing survives `workers(n)` changes.
#[derive(Clone)]
struct Invoices {
    publishers: TransactionalPartitions,
}

impl Invoices {
    async fn issue(&self, order: &Order, partition: i32) -> Result<(), KafkaError> {
        let publisher = self.publishers.for_partition(partition);
        publisher.begin_transaction().await?;
        for item in &order.items {
            let line = ItemShipment {
                order_id: order.id,
                item: item.clone(),
            };
            let payload = JsonCodec.encode(&line).expect("serializable");
            let outgoing = OutgoingMessage::new("invoice-lines", payload.as_ref());
            if let Err(err) = publisher.publish(outgoing).await {
                publisher.abort().await.ok();
                return Err(err);
            }
        }
        publisher.commit().await
    }
}

#[subscriber(
    KafkaTopic::new("billing").group("billing-svc").commit(Commit::Tracked),
    workers(4, by_key)
)]
async fn bill(
    order: &Order,
    // The delivery's source partition picks the lane's publisher; the key injects it as a
    // plain argument (the DI form of `ctx.context(keys::Partition)`).
    Ctx(partition): Ctx<Partition>,
    State(invoices): State<Invoices>,
) -> HandlerResult {
    if invoices.issue(order, partition).await.is_err() {
        return HandlerResult::retry();
    }
    HandlerResult::Ack
}

This deliberately does not compose with LaneKey::RecordKey pools: record-key lanes spread one partition across lanes, two lanes would collide on its publisher, and a per-lane id scheme would tie fencing identity to a runtime knob (n) instead of a Kafka-native unit. Sharing one id across a pool is the exactly-once pipeline below.

Exactly-once pipelines

EosPipeline is the full consume-transform-produce shape (KIP-447): one transactional producer shared by every lane, committing the consumed offsets inside the transaction (send_offsets_to_transaction), so source positions move atomically with the published records. A crash or an aborted window rewinds both - handlers reprocess (at-least-once on the handler side, as always), but the output topic never sees a duplicate.

Three places name one id:

// Exactly-once: every lane publishes into one shared EosPipeline, and the pipeline commits
// the consumed offsets inside the producer transaction (send_offsets_to_transaction), so
// source positions move atomically with the published records. A crash or an aborted window
// rewinds both - the output topic never sees a duplicate. The subscription's
// `Commit::Transactional` names the pipeline id; the consumer stops committing on its own.
//
// A publishing handler just returns the value: the runtime encodes it, and the pipeline's
// reply publisher (wired below) pairs it with this delivery's consumed offset automatically.
#[subscriber(
    KafkaTopic::new("raw-orders")
        .group("enrich-svc")
        .commit(Commit::Transactional("enrich-svc-1".into())),
    publish("enriched-orders"),
    workers(4, by_key)
)]
async fn enrich(order: &Order) -> Order {
    order.clone()
}

A publishing handler needs no manual pairing at all: mount it with the pipeline's reply publisher, and every reply joins the window paired with its delivery's consumed offset -

// Every reply of `enrich` rides the pipeline's window, paired with its offset.
b.include_publishing(enrich, pipeline.replies());

pipeline.replies() is a plain TypedPublisher under the hood (the explicit spelling is TypedPublisher::new(pipeline.clone()).transform(EosReplies)), so codecs and further transforms compose as usual; replies_with(codec) names a non-default codec. For manual publishes from a plain handler, EosPipeline::publish takes the delivery's coordinates explicitly - as a Ctx<Source> extractor parameter, like every other KafkaContext field key.

The subscription's Commit::Transactional("enrich-svc-1") switches its consumer's own committing off (the pipeline owns the offsets) and registers its watermark with the pipeline; EosPipeline::new(broker.publisher().transactional_id("enrich-svc-1")) wires the producer side. Publishes join the pipeline's open window; every commit_interval (100ms by default, the Kafka Streams EOS default) the window closes: the pipeline waits for its participants to settle, adds the settled positions and the consumer's group metadata to the transaction, and commits. The group metadata is what fences a stale consumer server-side, so a rebalance mid-window makes the commit fail instead of committing offsets the consumer no longer owns.

On any failure - a failed publish, a commit error, or a settle stall (a handler hanging or retry()-ing past the publisher's transaction timeout) - the window aborts and the consumers seek back to the last committed offsets, so the whole window redelivers promptly and republishes into a fresh transaction. Records published into an aborted window were never visible to read_committed readers (librdkafka's default here).

Practical notes:

  • One pipeline id per service instance, exactly like any transactional id: it is the fencing unit. Kafka Streams' EOSv2 uses the same one-producer-per-process scheme.
  • End-to-end latency is at least the commit interval: records become visible at the window commit, not at publish.
  • retry() from a participant stalls its window until the transaction deadline, then aborts it; prefer drop()/dead-lettering for poison messages in EOS handlers.
  • The retry_after/retry_via deferred-republish fallback does not apply to EOS replies: a delayed copy would break the offset-record pairing.
  • The reply publisher pairs only with Commit::Transactional subscriptions naming this pipeline's id; a reply from any other subscription fails with a clear error instead of silently downgrading the guarantee.
  • Works best over the default LaneKey::Partition lanes, where each partition settles in order behind its lane head.

Back-pressure and shutdown

A publish waits indefinitely for space when librdkafka's local queue is full - the natural back-pressure behavior. KafkaPublisher::queue_timeout bounds that wait instead, failing the publish with a queue-full error.

Broker::shutdown flushes in-flight publishes and reports an error when they do not make it out within KafkaBroker::flush_timeout (30 seconds unless configured).