Keys

The partition-key header, PARTITION_KEY_HEADER, becomes the record's native key, so records sharing a key land on one partition, in order.

The header is named kafka-partition-key. On publish its value becomes the Kafka record key and does not travel as a header of its own. Kafka's partitioner hashes the key, so every record of one key goes to one partition, and a consumer reads them in the order they were written. A message type declares the header in its typed header contract, #[outgoing(headers = ..)], with a field renamed to kafka-partition-key; a HeaderMap with that entry works the same. On consume the header holds the record key again, and a handler reads the key as raw bytes through the Key context key. workers(n, by_key) with LaneKey::RecordKey keeps the same order inside one consumer: records of one key run on one lane. A partition step places a record ahead of its key.

Example

Macros
Manual

examples/kafka_publishing_keys.rs
/// The header a shipment carries: `kafka-partition-key`, which becomes the record key.
#[derive(Debug, PartialEq, Deserialize, Serialize)]
struct ShipmentKey {
    #[serde(rename = "kafka-partition-key")]
    customer: String,
}

/// Every shipment of one customer lands on one partition, so its consumer sees them in order.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "shipments", headers = ShipmentKey)]
struct Shipment {
    order_id: u64,
}

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

#[subscriber(KafkaTopic::new("orders").group("shipping"))]
async fn ship(order: &Order, Out(shipments): Out<impl Publisher, Shipments>) -> HandlerOutcome {
    let key = ShipmentKey {
        customer: order.customer.clone(),
    };
    let shipment = Shipment { order_id: order.id };
    if shipments
        .message(&shipment)
        .with_headers(&key)
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/kafka_publishing_keys.rs
/// The header a shipment carries: `kafka-partition-key`, which becomes the record key.
#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct ShipmentKey {
    #[serde(rename = "kafka-partition-key")]
    customer: String,
}

/// Every shipment of one customer lands on one partition, so its consumer sees them in order.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "shipments", headers = ShipmentKey)]
struct Shipment {
    order_id: u64,
}

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

struct Ship;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Ship
where
    Entry: OutEntry<Shipments, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let key = ShipmentKey {
            customer: order.customer.clone(),
        };
        let shipment = Shipment { order_id: order.id };
        outs.get(Shipments)
            .message(&shipment)
            .with_headers(&key)
            .publish()
            .await
            .map_err(|_| HandlerOutcome::retry())
    }
}

fn orders() -> KafkaTopic {
    KafkaTopic::new("orders").group("shipping")
}

Example​

Example