partition

The partition step pins one record to a partition, ahead of its key and the partitioner.

It is a per-record setting of Kafka's publisher, KafkaOptions, written by a step of the publish builder. A handler reaches the step through a slot bounded on those options: Out<impl Publisher<Options = KafkaOptions>, Slot>, with this crate's prelude imported. The bound keeps the step off a publisher of another broker. The record goes to exactly that partition; its key still travels with it. A partition the topic does not have fails the publish, and the handler decides what follows. A test reads the setting back from the slot with tb.out::<Slot>().with_options(..). The in-process cluster gives every topic one partition, so a pin to partition 0 lands and a pin to any other is refused, as a cluster refuses a partition its topic lacks.

Example

Macros
Manual

examples/kafka_publishing_partition.rs
/// Each partition of `dispatch.lanes` is one courier lane, read by that lane's courier alone, so
/// a pickup goes to the partition its order names.
#[subscriber(KafkaTopic::new("dispatch.orders").group("dispatch"))]
async fn dispatch(
    order: &Order,
    Out(lanes): Out<impl Publisher<Options = KafkaOptions>, Lanes>,
) -> HandlerOutcome {
    let pickup = Pickup { order_id: order.id };
    if let Err(err) = lanes.message(&pickup).partition(order.lane).publish().await {
        tracing::warn!(order.id, order.lane, %err, "pickup not dispatched");
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/kafka_publishing_partition.rs
/// Each partition of `dispatch.lanes` is one courier lane, read by that lane's courier alone, so
/// a pickup goes to the partition its order names.
struct Dispatch;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Dispatch
where
    Entry: OutEntry<Lanes, Wire: Publisher<Options = KafkaOptions>>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let pickup = Pickup { order_id: order.id };
        let sent = outs
            .get(Lanes)
            .message(&pickup)
            .partition(order.lane)
            .publish()
            .await;
        if let Err(err) = sent {
            tracing::warn!(order.id, order.lane, %err, "pickup not dispatched");
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

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

Example​

Example