RoundRobin

RoundRobin spreads replies over a fixed partition count, placing only records with no key and no partition.

librdkafka places a keyless record by its partitioner, which may keep a whole batch on one partition. When every message takes about the same long time to process, that leaves one consumer busy and its peers idle. RoundRobin::partitions(n) pins each record to the next partition of the cycle 0, 1, ..., n - 1, one record each. It is a transform of the publish position, written after the policy: .out_reply(Publish::default()).transform(RoundRobin::partitions(n)). It writes the same setting as the partition step, so it mounts over Kafka's publisher only. A record that carries a key or already names a partition keeps its placement. The count is fixed at the mount site and matches the destination topic: a smaller one leaves the last partitions empty, a larger one makes publishes fail.

Example

Macros

examples/kafka_distribution.rs

/// The service on the cluster `broker` connects to.
fn service(broker: KafkaBroker) -> impl App {
    RustStream::new(AppInfo::new("planner", "0.1.0")).with_broker(broker, |b| {
        // Every reply targets the next partition of the cycle (0, 1, ..., 7, 0, ...), one
        // record each. A reply that already names a partition or carries a record key keeps
        // its placement. The count matches the partitions of `planner.work-items`.
        b.include(plan)
            .out_reply(Publish::default())
            .transform(RoundRobin::partitions(8));
    })
}

Manual

examples/manual/kafka_distribution.rs

/// The service on the cluster `broker` connects to.
fn service(broker: KafkaBroker) -> impl App {
    RustStream::new(AppInfo::new("planner", "0.1.0")).with_broker(broker, |b| {
        let plan = subscriber("planner.orders", Plan).map_source(|_| orders());
        // Every reply targets the next partition of the cycle (0, 1, ..., 7, 0, ...), one
        // record each. A reply that already names a partition or carries a record key keeps
        // its placement. The count matches the partitions of `planner.work-items`.
        b.include(plan.reply().build())
            .out_reply(Publish::default())
            .transform(RoundRobin::partitions(8));
    })
}