KafkaPartitions

KafkaPartitions reads exactly the partitions it names, joins no group and never rebalances.

KafkaPartitions::new takes the topic and its partition numbers. It suits a reader pinned to a partition, an inspection or replay tool, and one consumer per partition. The reader takes no part in the group protocol, so group members on the same topic neither see it nor take its partitions away. A group stays optional and decides only where offsets go. With group the reader commits into that group without joining it. Without a group it commits nothing, so start must name Earliest or Latest: Committed has nothing to resume from and stops the service at startup. commit, lane_key and config work as on KafkaTopic; Commit::Transactional stops the service at startup, because an exactly-once pipeline commits through the group protocol. Under the default lane key each named partition gets a lane of its own. The subscription's name is the topic. No publish addresses a subset of partitions, so a registration names where a retry copy goes, as the example does with .out_retry(policy).to("orders.retry").

Example

Partition 0 of orders, read from its first record by a service that names no group.

Macros
Manual

examples/kafka_subscriptions_partitions.rs
/// Inspects partition 0 of `orders` from its first record, with no group of its own.
#[subscriber(KafkaPartitions::new("orders", [0]).start(StartOffset::Earliest))]
async fn inspect(order: &Order, Ctx(offset): Ctx<Offset>) -> HandlerOutcome {
    tracing::info!(order.id, offset, "partition 0");
    HandlerOutcome::ack()
}

fn app() -> impl App {
    // The broker names no group, and the reader needs none.
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("inspector", "0.1.0")).with_broker(broker, |b| {
        // No publish reaches exactly these partitions, so a retried order goes to a topic of
        // its own.
        b.include(inspect)
            .out_retry(Publish::default())
            .to("orders.retry");
    })
}

examples/manual/kafka_subscriptions_partitions.rs
/// Logs each order with its offset in the partition.
struct Inspect;

impl Handle<Order, (), (), KafkaContext> for Inspect {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, KafkaContext>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, offset = ctx.context(Offset), "partition 0");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    // The broker names no group, and the reader needs none.
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("inspector", "0.1.0")).with_broker(broker, |b| {
        // Partition 0 of `orders`, from its first record.
        let partition = KafkaPartitions::new("orders", [0]).start(StartOffset::Earliest);
        // No publish reaches exactly these partitions, so a retried order goes to a topic of
        // its own.
        b.include(
            subscriber("orders", Inspect)
                .build()
                .map_source(|_| partition),
        )
        .out_retry(Publish::default())
        .to("orders.retry");
    })
}

Example​

Example