LaneKey

LaneKey decides what workers(n, by_key) keys a lane by: the source partition, or the record key.

workers(n, by_key) runs n sequential lanes: deliveries with one lane key run in order on one lane, and different keys run at once. On Kafka the descriptor's lane_key decides what that key is. LaneKey::Partition, the default, keys a lane by the partition the record came from, Kafka's own unit of order. Everything one partition delivers then runs in order, keyless records included, and partitions run side by side. LaneKey::RecordKey keys a lane by the record's native key, which is finer than a partition. One key's records stay in order, and records of different keys run at once even inside one partition. A keyless record has no lane key under RecordKey; it rotates over the lanes and loses its partition order. The publish side sets the record key from the partition-key header, so a record published with a key reaches its lane by that key. The manual form writes the lanes as .workers_by_key(n) on the mount chain.

Example

Two tenants' orders, each order a second of work: lanes by record key run the tenants side by side, two seconds in all.

Macros

examples/kafka_keys.rs

/// Eight lanes by record key: one tenant's orders run in order, different tenants at once,
/// even inside one partition.
#[subscriber(
    KafkaTopic::new("orders").group("orders-workers").lane_key(LaneKey::RecordKey),
    workers(8, by_key)
)]
async fn on_order(order: &Order) -> HandlerOutcome {
    process(order).await;
    HandlerOutcome::ack()
}

Manual

examples/manual/kafka_keys.rs

/// Eight lanes by record key: one tenant's orders run in order, different tenants at once,
/// even inside one partition.
fn app() -> impl App<State = ()> {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        let orders = KafkaTopic::new("orders")
            .group("orders-workers")
            .lane_key(LaneKey::RecordKey);
        b.include(
            subscriber("orders", Process)
                .workers_by_key(nonzero!(8))
                .build()
                .map_source(|_| orders),
        );
    })
}

Lanes by partition

The same four orders on a topic with one partition take four seconds: the partition is one lane.

Macros

examples/kafka_keys.rs

/// Eight lanes by source partition, the default: everything one partition delivers runs in
/// order on one lane, keyless records included.
#[subscriber(KafkaTopic::new("audit").group("audit-workers"), workers(8, by_key))]
async fn on_audit(order: &Order) -> HandlerOutcome {
    process(order).await;
    HandlerOutcome::ack()
}

Manual

examples/manual/kafka_keys.rs

/// Eight lanes by source partition, the default: everything one partition delivers runs in
/// order on one lane, keyless records included.
fn app() -> impl App<State = ()> {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(broker, |b| {
        let audit = KafkaTopic::new("audit").group("audit-workers");
        b.include(
            subscriber("audit", Process)
                .workers_by_key(nonzero!(8))
                .build()
                .map_source(|_| audit),
        );
    })
}