Keys

The partition-key header, PARTITION_KEY_HEADER, is the message key: it picks the partition of a partitioned topic and orders KeyShared and by_key lanes.

The header is named partition-key. On publish its value becomes the Pulsar message key and does not travel as a property of its own. On a partitioned topic the producer hashes the key, so every message of one key goes to one partition, in the order it was published. A publish without a key spreads over the partitions in turn. A message type declares the header in its typed contract, #[outgoing(headers = ..)], with a field renamed to partition-key; a HeaderMap with that entry works the same. On consume the key comes back as that header. A KeyShared subscription hands every message of one key to one consumer, and workers(n, by_key) keeps one key on one lane inside a consumer. A handler body that keys one publish of a slot uses the partition_key step instead.

Subscription types and ordering places keyed messages on the partitions of a partitioned topic.

Example

Macros
Manual

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

/// Every shipment of one customer carries one key, so the partition and the consumer that
/// take it see that customer's shipments 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(PulsarSubscription::new("orders", "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/pulsar_publishing_keys.rs
/// The header a shipment carries: `partition-key`, which becomes the message key.
#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
struct ShipmentKey {
    #[serde(rename = "partition-key")]
    customer: String,
}

/// Every shipment of one customer carries one key, so the partition and the consumer that
/// take it see that customer's shipments in order.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[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())
    }
}

Example​

Example