Assignment

Assignment picks the consumer group's rebalance strategy, eager or cooperative sticky.

A group hands each partition of its topics to one member, and hands them out again when a member joins or leaves. The setting is librdkafka's partition.assignment.strategy, written on the descriptor with assignment. Assignment::Range and Assignment::RoundRobin are eager: every member gives up all its partitions at each rebalance, and the group stops reading until the new assignment lands. Assignment::CooperativeSticky is incremental: only the partitions that change hands move, and the rest keep flowing during the rebalance. Without the setting librdkafka uses range,roundrobin. Every member of one group must use the same family: librdkafka refuses a cooperative member in an eager group, and the error surfaces on the subscription. KafkaPartitions joins no group and has no assignment.

Example

A cooperative sticky group reads every shipment once, whichever member holds its partition.

Macros

examples/kafka_subscriptions_assignment.rs

/// Ships each shipment; a rebalance moves only the partitions that change hands.
#[subscriber(
    KafkaTopic::new("warehouse.shipments")
        .start(StartOffset::Earliest)
        .assignment(Assignment::CooperativeSticky)
)]
async fn ship(shipment: &Shipment) -> HandlerOutcome {
    tracing::info!(shipment.id, "shipped");
    HandlerOutcome::ack()
}

Manual

examples/manual/kafka_subscriptions_assignment.rs

/// Ships each shipment it reads.
struct Ship;

impl Handle<Shipment> for Ship {
    fn handle(
        &self,
        shipment: &Shipment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(shipment.id, "shipped");
        ready(Ok(()))
    }
}

fn app(servers: &str, group: &str) -> impl App<State = ()> {
    let broker = KafkaBroker::new([servers]).default_group(group);
    RustStream::new(AppInfo::new("warehouse", "0.1.0")).with_broker(broker, |b| {
        // A rebalance moves only the partitions that change hands.
        let shipments = KafkaTopic::new("warehouse.shipments")
            .start(StartOffset::Earliest)
            .assignment(Assignment::CooperativeSticky);
        b.include(
            subscriber("warehouse.shipments", Ship)
                .build()
                .map_source(|_| shipments),
        );
    })
}